Cloud Provider SDKs in Python: Architecture & Implementation Patterns

A cloud provider SDK is the thinnest useful layer of infrastructure automation available to a Python engineer: boto3 for AWS, google-cloud-* for Google Cloud, azure-mgmt-* for Azure. Each is a generated client that signs an HTTPS request, sends it to a regional endpoint, and parses the response into a dictionary. There is no plan, no diff, no checkpoint file — every guarantee a Pulumi or CDKTF program hands you for free becomes your responsibility the moment you call the SDK directly. This topic sits inside Python IaC Fundamentals & Strategy and answers the two questions engineers hit first when the two layers meet in one repository. Best practices for managing cloud credentials in Python explains how each provider's credential chain resolves, how to keep long-lived keys out of the process, and how to type the loader so a missing variable fails at import rather than mid-deploy. Using boto3 inside Pulumi and CDKTF covers the narrow set of calls that are safe to issue from inside a framework program, and how to fence off the ones that are not so the engine's state never falls behind reality.

Problem Framing

The failure this topic prevents is a codebase where nobody can say which layer owns a given resource. A team starts with Pulumi, hits a resource the provider does not model yet — an AWS Service Quotas increase, a Route 53 domain registration, an Azure subscription-level feature flag — and drops to the SDK inside the program to do it imperatively. Six months later the SDK call still runs on every pulumi up, it is not idempotent, it has no preview representation, and a pulumi destroy leaves its side effect behind. The resource exists, costs money, and appears in no state file anywhere.

The reverse failure is just as common. A platform team writes a 900-line boto3 script that creates a VPC, subnets, route tables, and a NAT gateway, stores the resulting IDs in a JSON file next to the script, and treats that file as state. It works until two engineers run it concurrently, or until someone deletes a subnet in the console. The script has no concept of "the subnet should exist with these attributes" — only "call create_subnet" — so its second run either errors with InvalidSubnet.Conflict or silently creates a duplicate.

Both failures come from the same missing decision: which layer owns desired state. The SDK is an excellent instrument for asking questions and for one-shot actions that are not part of a resource's lifecycle. It is a poor instrument for maintaining a resource, because maintenance is exactly the set of concerns the SDK deliberately leaves out.

Read that grid as a checklist. Every row where the middle column says "you" is code you will write, test, and maintain. For a nightly tag-audit script that is a fair trade; for a production network it is not.

Prerequisites

Before working through the patterns below, have the following in place:

• Python 3.9 or newer with a project-local virtual environment. The typed examples use from __future__ import annotations semantics and standard-library generics, and Pulumi's Python runtime resolves imports against whichever interpreter is on PATH when the engine starts.
• Pinned SDK versions in a lockfile. boto3 and botocore ship near-daily releases whose service models change default behaviour; treat them as production dependencies, not as ambient tools. Dependency workflow is covered in Setting Up Dev Environments.
• A credential source that produces short-lived tokens — an instance profile, an OIDC-federated CI role, az login, or Workload Identity Federation. No example on this page reads a static access key.
• moto (in-process AWS mocking) or an equivalent local stub, so the test suite never touches a live account.

# CLI: bash scripts/bootstrap_sdk_env.sh
python3 -m venv .venv
.venv/bin/pip install "boto3==1.34.106" "botocore==1.34.106" "tenacity==8.3.0" \
  "moto[ec2,s3]==5.0.9" "pytest==8.2.1" "mypy==1.10.0"

# Verify the credential chain resolves to a role, not a static user.
# Provider note: an ARN containing :assumed-role/ means STS issued the session.
.venv/bin/python -c "import boto3; print(boto3.client('sts').get_caller_identity()['Arn'])"

If that last command prints an ARN of the form arn:aws:iam::123456789012:user/deploy, stop and fix the credential source before continuing — every pattern below assumes a session that expires.

Architecting Provider SDK Integration for IaC Workflows

Transitioning from declarative HCL to programmatic SDK calls requires strict execution boundary definitions. Align your runtime strategy with the Python IaC Fundamentals & Strategy framework to prevent implicit state leakage. Evaluate trade-offs between raw API calls (boto3, google-cloud-, azure-mgmt-) and higher-level framework abstractions (Pulumi, CDKTF) before committing to a control plane. The two recurring questions this section answers are how to authenticate clients without leaking keys — covered in depth by Best practices for managing cloud credentials in Python — and when it is safe to call the SDK from inside a framework program, covered by Using boto3 inside Pulumi and CDKTF.

Provider Client Instantiation & Region Routing

Initialize SDK clients using explicit region routing to prevent cross-region resource collisions. Configure connection pools and socket timeouts to mitigate transient network failures during bulk provisioning. Unrouted clients default to environment-configured regions, causing race conditions in multi-region deployments.

The mechanism is worth understanding precisely, because it is the single most common source of "it worked on my laptop" incidents. When you call boto3.client("ec2") with no region_name, botocore walks a resolution chain: the explicit argument, then AWS_REGION, then AWS_DEFAULT_REGION, then the region key of the active profile in ~/.aws/config, then the instance metadata service if one is reachable. A developer workstation usually resolves to whatever the engineer last worked in; a GitHub Actions runner has none of those set and raises botocore.exceptions.NoRegionError: You must specify a region. The same client construction therefore produces a call to ec2.eu-west-1.amazonaws.com locally and a hard failure in CI. Pass region_name explicitly on every client, derive it from typed configuration, and the chain never runs.

Connection settings matter for the same reason. botocore's defaults are a 60-second connect timeout and a 60-second read timeout with a pool of 10 connections. A provisioning script that fans out 40 describe_* calls through a ThreadPoolExecutor will queue on that pool and appear to hang; a script behind a broken NAT route will sit for a full minute per call before it reports anything. Setting connect_timeout=10 and max_pool_connections=32 on a botocore.config.Config object turns a silent stall into a fast, actionable error.

Dependency Pinning & Version Compatibility Matrix

Lock provider SDK versions to exact semantic releases to guarantee deterministic execution. Validate compatibility matrices against cloud API changelogs before upgrading core dependencies. Silent SDK drift alters default resource attributes, which can corrupt existing infrastructure state when Pulumi or CDKTF re-reads provider schemas.

boto3 is a thin façade over botocore, and botocore carries the JSON service models that define every operation, parameter, and default. Those models ship in patch releases. When AWS changed the default for S3 bucket encryption, a botocore upgrade changed what create_bucket produced without a single line of your code changing. Pin both packages to exact versions, upgrade them deliberately in their own pull request, and run the full test suite against the new models. A range specifier such as boto3>=1.34 guarantees only that two engineers on the same commit can produce different infrastructure.

The same discipline applies one level up. A Pulumi program depends on pulumi_aws, which embeds a specific Terraform AWS provider build; a CDKTF program depends on generated provider bindings tied to a terraform binary version. Bumping pulumi-aws from 6.x to 7.x can change a resource's default arguments and produce a diff on resources you did not touch. Read the provider changelog before the upgrade and run a preview against a non-production stack first.

Environment Bootstrapping & Toolchain Alignment

Standardize Python runtimes and virtual environment isolation across developer workstations and CI runners. Reference execution model differences outlined in Python vs Terraform vs Ansible to select appropriate orchestration layers.

The factory below combines the three concerns: an explicit region, a bounded timeout, and a retry policy that distinguishes between errors worth retrying and errors that will never succeed. Note that tenacity wraps the construction path here, while botocore's own retries block handles per-operation retries — adaptive mode reads the throttling response headers and slows the client down rather than hammering a rate-limited endpoint.

from typing import Protocol
import boto3
from botocore.config import Config
from botocore.exceptions import ClientError
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

class ProviderClient(Protocol):
    def describe_resource(self, **kwargs) -> dict: ...

@retry(
    stop=stop_after_attempt(4),
    wait=wait_exponential(multiplier=1, min=2, max=30),
    retry=retry_if_exception_type(ClientError),
    reraise=True,
)
def initialize_aws_client(region: str, timeout: int = 10) -> boto3.client:
    """Factory function for AWS SDK clients with exponential backoff on ClientError."""
    config = Config(
        retries={"max_attempts": 4, "mode": "adaptive"},
        connect_timeout=timeout,
        read_timeout=timeout,
        region_name=region,
    )
    return boto3.client("ec2", config=config)

# CLI: python -c "from infra.clients import initialize_aws_client; initialize_aws_client('eu-west-1')"
# Provider note: region_name inside Config overrides the AWS_REGION resolution chain entirely.
# pytest integration: Mock via @pytest.fixture returning MagicMock(spec=boto3.client)

One subtlety: retry_if_exception_type(ClientError) is deliberately broad for a factory whose only network call is metadata resolution, but it is the wrong policy for a mutating operation. ClientError covers AccessDenied, ValidationError, and ThrottlingException alike, and retrying an authorization failure four times only delays the report by thirty seconds. For mutations, inspect exc.response["Error"]["Code"] and retry only the transient set — ThrottlingException, RequestLimitExceeded, InternalError, ServiceUnavailable.

State Management & Resource Lifecycle Patterns

When using raw SDK calls outside a framework, you are responsible for your own state tracking. Align local development workflows using Setting Up Dev Environments to guarantee consistent SDK behavior across tiers. Bypassing native state managers (Pulumi backend, Terraform state) requires rigorous idempotency enforcement.

Custom State Serialization with JSON/YAML Backends

Persist resource identifiers and metadata to version-controlled JSON or YAML manifests. Implement optimistic concurrency control using ETags or revision tokens during write operations. Missing serialization locks allow concurrent executions to overwrite live resource mappings.

A local file is not a state backend. If the manifest lives on disk next to the script, two engineers running it at the same time both read the pre-change version and the second write silently discards the first. The minimum viable substitute is an S3 object with versioning enabled plus a conditional write: read the object, keep its ETag, and pass that ETag back on put_object so the write fails with PreconditionFailed when someone else has written in the meantime. On Google Cloud the equivalent is if_generation_match; on Azure Blob Storage it is an If-Match header carrying the blob's ETag. That is the same concurrency primitive Pulumi and Terraform implement, which is a good hint that if you are building it by hand you may want the engine instead — the trade-offs are laid out in choosing a state backend for Python IaC.

Idempotency Keys & Conditional Resource Provisioning

Generate deterministic keys from infrastructure parameters to prevent duplicate resource creation. Wrap SDK mutations in conditional logic that queries existing state before issuing create commands. Unchecked mutations trigger orphaned resources and quota exhaustion.

Several AWS APIs accept an explicit idempotency token and will do the deduplication for you: ec2:RunInstances takes ClientToken, ec2:CreateNatGateway takes ClientToken, and sqs:SendMessage takes MessageDeduplicationId for FIFO queues. A token is honoured for a bounded window — 24 hours for most EC2 operations — so a retry inside that window returns the original resource rather than creating a second one. Derive the token deterministically from the inputs (sha256 over the sorted parameter dictionary, truncated to 64 characters) and a retried run is genuinely safe. Where no token parameter exists, fall back to a tag-based lookup: tag every resource with a stable logical name at creation, and make the create path a describe_* filtered on that tag first.

Drift Detection via SDK Read-Only Polling

Schedule periodic read-only API scans to compare live cloud configurations against serialized manifests. Emit structured telemetry when attribute divergence exceeds acceptable thresholds. Unmonitored drift invalidates deployment assumptions and breaks rollback procedures.

Polling has one trap worth naming: cloud APIs return normalized values, not the values you sent. Submit a security-group rule with CIDR 10.0.0.0/8 and describe_security_groups returns it unchanged; submit an S3 bucket policy and get_bucket_policy returns a re-serialized JSON document with reordered keys and different whitespace. A naive string comparison reports drift on every poll. Compare parsed structures with a canonical ordering, and exclude provider-managed fields — CreationDate, OwnerId, LastModifiedTime — from the comparison set. The same normalization problem, and the refresh-based way frameworks solve it, is covered in detecting and remediating state drift in Python IaC.

import json
from pathlib import Path
from typing import Any, Dict, Optional

class IdempotentProvisioner:
    """Context manager ensuring safe resource creation via pre-flight existence checks."""

    def __init__(
        self,
        state_path: Path,
        client: Any,
        resource_type: str,
        params: Dict[str, Any],
    ) -> None:
        self.state_path = state_path
        self.client = client
        self.resource_type = resource_type
        self.params = params
        self.resource_id: Optional[str] = None

    def __enter__(self) -> "IdempotentProvisioner":
        self._load_state()
        self._ensure_resource()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
        if exc_type is None:
            self._persist_state()

    def _load_state(self) -> None:
        if self.state_path.exists():
            state = json.loads(self.state_path.read_text())
            self.resource_id = state.get(self.resource_type)

    def _ensure_resource(self) -> None:
        if self.resource_id:
            return  # Already exists—skip creation
        response = self.client.create_resource(**self.params)
        self.resource_id = response["ResourceId"]

    def _persist_state(self) -> None:
        state = {self.resource_type: self.resource_id}
        self.state_path.write_text(json.dumps(state, indent=2))

# CLI: python -m infra.provision --state ./state/network.json --resource vpc
# State implication: _persist_state runs only on a clean exit, so a crash mid-create
# leaves the resource live but unrecorded — the classic orphan.
# pytest integration: Mock client.create_resource and assert _ensure_resource
# skips creation on the second call (resource_id already populated from state).

That last comment names the honest limitation of any hand-rolled state layer: the window between the provider acknowledging a create and your process recording the identifier. Pulumi closes it by writing a pending-operation marker to the checkpoint before the RPC and reconciling on the next run; a script closes it either by tagging the resource with its logical name at creation (so a rescan can adopt it) or by accepting a manual cleanup step. Choose consciously, and document which one you chose.

Modularizing Infrastructure with Pulumi and CDKTF

Bridge raw SDK invocations to Pulumi and CDKTF resource models through strict abstraction boundaries. Package provider-specific logic into versioned Python libraries that expose uniform interfaces. Maintain native SDK performance characteristics by minimizing framework overhead during graph resolution.

Component Resource Abstraction Layers

Encapsulate provider-specific API calls behind interface-compliant Python classes. Expose standardized methods for provisioning, updating, and tearing down infrastructure primitives. Leaky abstractions expose raw API errors, complicating framework-level error handling.

In Pulumi the natural boundary is pulumi.ComponentResource. A component owns its children, registers a single URN in state, and can expose SDK-derived facts as outputs — but only through pulumi.Output.from_input or an apply, never as a bare Python value read at construction time. A component that calls boto3 in __init__ and stores the result as a plain string bakes a point-in-time answer into the program: it is evaluated during preview, it is not tracked, and it will not change when the underlying resource changes. In CDKTF the equivalent boundary is a Construct subclass, and the same rule applies — a value fetched by boto3 at synth time is frozen into the emitted cdk.tf.json, visible in the diff as a hard-coded literal.

Stack-Level Configuration & Parameter Injection

Inject environment-specific parameters via typed configuration objects rather than raw environment variables. Validate parameter schemas at import time to fail fast before graph evaluation begins. Late-binding configuration causes partial deployments and inconsistent resource tagging.

Concretely: read pulumi.Config() once at the top of __main__.py, feed the values into a Pydantic model, and pass that model into every component constructor. If a required key is missing, Pulumi raises pulumi.errors.RunError: Missing required configuration variable 'aws:region' before a single resource is registered, and the stack is untouched. If instead each component reads os.environ on demand, a missing key surfaces halfway through an update with several resources already created — a partial deployment that must be reconciled by hand.

Cross-Provider Dependency Graphs

Resolve implicit dependencies by explicitly passing resource outputs between provider modules. Construct directed acyclic graphs using framework-native dependency managers. Circular references trigger infinite evaluation loops and corrupt deployment ordering.

An SDK call sitting between two resources is invisible to that graph. If a boto3 lookup needs an ID that another resource in the same program produces, the call must run inside an Output.apply so the engine schedules it after the dependency resolves — and even then the result is only available inside the callback, never as a synchronous return value. Attempting to read it eagerly yields the string Calling __str__ on an Output[T] is not supported, which is Pulumi telling you the value does not exist yet. Cross-provider work — an AWS resource whose ID feeds a Cloudflare record, say — should pass outputs explicitly rather than re-querying, because a re-query reintroduces the ordering problem the graph already solved.

Testing, Validation, and CI/CD Integration

Enforce rigorous unit and integration testing strategies before promoting SDK-driven infrastructure code. Mock provider responses to validate resource schemas and enforce policy gates during pipeline execution.

Unit Testing with moto/localstack and pytest

Isolate SDK calls using moto (intercepted in-process) or localstack (runs locally as a Docker container) to simulate cloud APIs without network egress. Parameterize test suites across multiple regions to validate routing and error handling. Live API testing during CI introduces flaky builds and unpredictable state mutations.

moto works by patching botocore's endpoint resolver, which has one consequence that catches everyone once: a client created before the mock_aws decorator takes effect still points at the real endpoint. Construct clients inside the test function or inside a fixture that depends on the mock, never at module import. Set the four AWS_* environment variables to dummy values in a session-scoped autouse fixture as well, so a stray unmocked call fails loudly with InvalidClientTokenId instead of quietly reaching a real account with a developer's ambient credentials. Deeper coverage of that setup lives in mocking AWS services with moto in pytest.

Also test the error paths, not only the happy path. moto cannot produce a ThrottlingException on demand, so use botocore.stub.Stubber to inject one and assert that your retry wrapper backs off the expected number of times before re-raising. A retry policy that has never been exercised is a retry policy that does not work.

Integration Testing Against Ephemeral Environments

Provision isolated cloud accounts or namespaces for end-to-end validation of complex resource graphs. Automate teardown routines to prevent resource accumulation and cost leakage. Persistent test environments accumulate orphaned state, skewing drift detection metrics.

Policy-as-Code Validation with Open Policy Agent (OPA)

Serialize resource configurations to JSON and evaluate against Rego policies before deployment. Block pipeline progression on policy violations to enforce compliance baselines. Post-deployment policy checks require manual remediation and increase blast radius.

import pytest
from unittest.mock import MagicMock
from pathlib import Path
import json
from infra.state import IdempotentProvisioner

@pytest.fixture
def mock_aws_client():
    client = MagicMock()
    client.create_resource.return_value = {"ResourceId": "res-12345"}
    return client

@pytest.fixture
def temp_state_path(tmp_path: Path) -> Path:
    return tmp_path / "test_state.json"

@pytest.mark.parametrize("region", ["us-east-1", "eu-west-1"])
def test_idempotent_provisioner_skips_existing(
    region: str, mock_aws_client: MagicMock, temp_state_path: Path
) -> None:
    # Arrange: Pre-populate state to simulate existing resource
    temp_state_path.write_text(json.dumps({"ec2_instance": "res-99999"}))

    # Act: Execute provisioner
    with IdempotentProvisioner(
        temp_state_path, mock_aws_client, "ec2_instance", {"region": region}
    ) as prov:
        assert prov.resource_id == "res-99999"

    # Assert: Create was never called because resource_id was loaded from state
    mock_aws_client.create_resource.assert_not_called()

# CLI: pytest tests/test_provisioner.py -v --cov=infra --tb=short
# CI integration: the same command runs as the `unit` job before any deploy job.

Security Hardening & Credential Orchestration

Secure SDK authentication flows by enforcing least-privilege access and automated credential rotation. Align runtime secrets management with Best practices for managing cloud credentials in Python to eliminate static key exposure.

Dynamic Credential Resolution via OIDC & STS

Configure SDKs to resolve short-lived tokens via OpenID Connect and Security Token Service exchanges. Implement automatic token refresh routines to maintain uninterrupted API access during long-running deployments. Expired tokens halt provisioning mid-execution, leaving resources in inconsistent states.

The refresh behaviour differs by how the session was created, and the difference decides whether a long deployment survives. A boto3.Session backed by an instance profile or by AssumeRoleWithWebIdentity refreshes automatically: botocore stores a RefreshableCredentials object that re-fetches roughly five minutes before expiry. A session built from the literal strings returned by an explicit sts.assume_role() call does not — those are frozen credentials, and when the hour is up every subsequent call fails with ExpiredToken: The security token included in the request is expired. If a deployment can run longer than the session duration, either raise DurationSeconds (up to the role's MaxSessionDuration, default 3600 seconds) or use the refreshable path and let botocore handle it.

Network Isolation & Private Endpoint Routing

Route SDK traffic through VPC endpoints or private service networks to bypass public internet exposure. Enforce DNS resolution policies that restrict API calls to authorized regional endpoints.

Audit Logging & SDK Call Telemetry

Capture structured telemetry for every SDK invocation, including request IDs, latency, and error codes. Pipe logs to centralized SIEM systems for real-time anomaly detection and compliance reporting. Unlogged mutations prevent forensic analysis during security incidents.

The request ID is the part that matters operationally. Every AWS response carries one in response["ResponseMetadata"]["RequestId"], and every ClientError exposes the same field on exc.response. Log it alongside the operation name and it becomes the join key between your run and the provider's own CloudTrail record — without it, support cases and post-incident timelines are guesswork. Register a botocore event hook (session.register("after-call", handler)) once at start-up rather than wrapping every call site.

Step-by-Step: A Typed, Auditable Client Layer

The patterns above converge on one small module that every script and every framework program imports. Build it in three steps.

1. Model the target account as typed configuration

The inputs to a client are a role to assume, a region, and a session name that shows up in CloudTrail. Model them once so a typo is a validation error rather than a 403 an hour later.

# infra/targets.py
# CLI: python -m infra.targets --check
from __future__ import annotations
from dataclasses import dataclass
from typing import Final
import re

_ROLE_ARN: Final = re.compile(r"^arn:aws:iam::\d{12}:role/[\w+=,.@-]+$")

@dataclass(frozen=True, slots=True)
class DeployTarget:
    """One account/region pair the automation is allowed to touch."""
    account_id: str
    region: str
    role_arn: str
    session_name: str = "python-iac"

    def __post_init__(self) -> None:
        if not _ROLE_ARN.match(self.role_arn):
            raise ValueError(f"role_arn is not a role ARN: {self.role_arn!r}")
        if not self.region.count("-") == 2:
            raise ValueError(f"region looks malformed: {self.region!r}")

# Provider note: session_name appears verbatim in the CloudTrail userIdentity record,
# so keep it stable and greppable across runs.

2. Broker every client through one factory

The factory is the only place in the codebase that calls assume_role, the only place that sets timeouts, and the only place that decides retry policy. Everything else asks it for a client.

# infra/clients.py
# CLI: python -c "from infra.clients import ClientFactory; print(ClientFactory(t).client('ec2'))"
from __future__ import annotations
from typing import Any
import boto3
from botocore.config import Config
from botocore.credentials import RefreshableCredentials
from botocore.session import get_session
from infra.targets import DeployTarget

class ClientFactory:
    """Builds region-pinned, role-scoped clients with refreshable credentials."""

    def __init__(self, target: DeployTarget) -> None:
        self._target = target
        self._config = Config(
            region_name=target.region,
            connect_timeout=10,
            read_timeout=30,
            max_pool_connections=32,
            retries={"max_attempts": 5, "mode": "adaptive"},
            user_agent_extra=f"python-iac/{target.session_name}",
        )
        self._session = self._build_session()

    def _build_session(self) -> boto3.Session:
        botocore_session = get_session()
        # Provider note: the assume-role provider in the default chain refreshes
        # automatically; hand-built frozen credentials do not.
        botocore_session.set_config_variable("region", self._target.region)
        creds = botocore_session.get_credentials()
        # DeferredRefreshableCredentials (assume-role profiles) subclasses this,
        # so one isinstance check covers instance profiles and OIDC sessions too.
        if not isinstance(creds, RefreshableCredentials):
            # Static keys reached the process — fail before anything is created.
            raise RuntimeError(
                "refusing to run with non-refreshable credentials; "
                "use an instance profile, OIDC role, or profile with role_arn"
            )
        return boto3.Session(botocore_session=botocore_session)

    def client(self, service: str) -> Any:
        return self._session.client(service, config=self._config)

# State implication: this module never mutates anything, so it is safe to import
# from a Pulumi program — no resource is created as a side effect of construction.

3. Restrict the call surface

A factory that hands out unrestricted clients invites the imperative-mutation problem back in. Wrap the operations the codebase actually needs and refuse the rest, so a reviewer can see the whole blast radius in one file.

# infra/lookups.py
# CLI: python -m infra.lookups --vpc-tag platform-core
from __future__ import annotations
from typing import Any
from botocore.exceptions import ClientError
from infra.clients import ClientFactory

class VpcLookup:
    """Read-only queries used to resolve pre-existing network facts."""

    def __init__(self, factory: ClientFactory) -> None:
        self._ec2 = factory.client("ec2")

    def id_by_name_tag(self, name: str) -> str:
        try:
            pages = self._ec2.get_paginator("describe_vpcs").paginate(
                Filters=[{"Name": "tag:Name", "Values": [name]}]
            )
            vpcs: list[dict[str, Any]] = [v for p in pages for v in p["Vpcs"]]
        except ClientError as exc:
            code = exc.response["Error"]["Code"]
            rid = exc.response["ResponseMetadata"]["RequestId"]
            raise RuntimeError(f"describe_vpcs failed ({code}, request {rid})") from exc
        if len(vpcs) != 1:
            raise LookupError(f"expected exactly one VPC tagged {name!r}, found {len(vpcs)}")
        return vpcs[0]["VpcId"]

# State implication: no create/modify/delete call exists on this class, so nothing
# here can drift the Pulumi checkpoint out of sync with the account.

The paginator matters. describe_vpcs returns at most 1000 results per page and describe_instances far fewer; a bare call that ignores NextToken silently reports a partial answer, and a lookup that returns the wrong ID is worse than one that fails.

Verification

Verification for an SDK layer has three parts: prove the credentials are what you think they are, prove the calls are read-only, and prove the retry path works.

# CLI: bash scripts/verify_sdk_layer.sh

# 1. Identity — must be an assumed role, and must be the expected account.
.venv/bin/python - <<'EOF'
import boto3
ident = boto3.client("sts").get_caller_identity()
assert ":assumed-role/" in ident["Arn"], f"not a role session: {ident['Arn']}"
print(ident["Account"], ident["Arn"])
EOF

# 2. Static analysis — no mutating verb may appear outside infra/mutations.py.
grep -REn "\.(create|delete|modify|put|terminate)_[a-z_]+\(" infra/ \
  --include="*.py" | grep -v "^infra/mutations.py" && exit 1

# 3. Unit tests, including the throttling path driven by botocore's Stubber.
.venv/bin/pytest tests/ -q
.venv/bin/mypy --strict infra/

The grep in step 2 is crude and that is the point: it runs in under a second, it is obvious to a reviewer, and it fails the build the day someone adds a create_bucket call to a module named lookups.py. Pair it with a CI job that runs the full suite against moto so no verification step needs a live account.

For the framework side, the check is that the SDK layer changed nothing. Run pulumi preview --diff before and after introducing the module; the two outputs must be byte-identical. If they differ, something in the import path is registering a resource or freezing a fetched value into the program.

Troubleshooting

Almost every SDK failure arrives as one exception type — botocore.exceptions.ClientError — carrying a service-specific code in exc.response["Error"]["Code"]. Triage starts by reading that code, not the message.

NoRegionError: You must specify a region.

Cause. A client was constructed without region_name in an environment where none of AWS_REGION, AWS_DEFAULT_REGION, or a profile default is set. It is the standard CI failure, because runners have a clean environment while workstations do not.

Fix. Pass the region explicitly through the typed DeployTarget and into Config(region_name=...). Do not paper over it by exporting AWS_DEFAULT_REGION in the pipeline — that reintroduces the environment dependency the typed config exists to remove.

ExpiredToken: The security token included in the request is expired

Cause. The session was built from frozen credentials returned by an explicit sts.assume_role() call, and the deployment outran the token's lifetime — one hour by default. Long applies over many resources hit this routinely.

Fix. Use a refreshable credential provider (instance profile, AssumeRoleWithWebIdentity, or a profile with role_arn in ~/.aws/config) so botocore renews the session transparently. If the role must be assumed explicitly, raise DurationSeconds and confirm the role's MaxSessionDuration allows it — requesting more than the role permits fails with ValidationError: The requested DurationSeconds exceeds the MaxSessionDuration set for this role.

ThrottlingException: Rate exceeded during a bulk run

Cause. A ThreadPoolExecutor fanned out more concurrent describes than the service's per-account request rate allows. EC2 describes and IAM in particular have low, undocumented, account-wide limits.

Fix. Set retries={"mode": "adaptive", "max_attempts": 5} on the client Config so botocore's client-side rate limiter reads the throttling signal and slows down, and cap the pool at a modest width — eight workers is usually plenty. Prefer one paginated call over N single-item calls: describe_instances with a filter costs one request where a loop over instance IDs costs one per instance.

An error occurred (AccessDenied) when calling the DescribeVpcs operation

Cause. The assumed role's policy is missing the action, or a permissions boundary or Service Control Policy is denying it above the role. The message is identical in both cases, which is why the request ID matters.

Fix. Run aws sts get-caller-identity to confirm which principal is actually in use, then simulate the call with aws iam simulate-principal-policy --policy-source-arn <role-arn> --action-names ec2:DescribeVpcs. If the simulation allows it but the live call denies it, the denial is coming from an SCP or boundary — check the organization account rather than the role. The least-privilege patterns in enforcing IAM least privilege in Python IaC cover writing the narrow policy that fixes this properly.

Calling __str__ on an Output[T] is not supported from a boto3 call in a Pulumi program

Cause. An SDK call was given a value that the engine has not resolved yet — typically vpc.id passed straight into a Filters list. During preview that value does not exist, so Pulumi refuses to stringify it.

Fix. Move the call inside pulumi.Output.all(vpc.id).apply(lambda args: lookup(args[0])), and guard the body with if pulumi.runtime.is_dry_run(): return None so preview does not issue a live request. Better still, avoid the lookup entirely and pass the output through as a component argument.

Key Takeaways

Raw cloud SDKs give Python engineers the flexibility to build precisely scoped automation, but that flexibility comes with the burden of idempotency and state tracking that frameworks like Pulumi and CDKTF provide for free. Use raw SDKs for tooling, automation scripts, and integration tests. Use Pulumi or CDKTF for resource lifecycle management where drift detection, state rollback, and plan previews are required.

FAQ

Do I need cloud SDKs if I use Pulumi or CDKTF?

Mostly no for provisioning, but SDKs like boto3 fill gaps — lookups and imperative actions — as shown in using boto3 inside Pulumi and CDKTF.

How should credentials be supplied?

Through the environment or an execution role, never hard-coded — the practices in managing cloud credentials apply to SDKs and providers alike.

Which SDK for which cloud?

boto3 for AWS, google-cloud-* for GCP, azure-* for Azure; each mirrors its provider's resource model closely enough to be a useful escape hatch. Google's libraries are split per service (google-cloud-storage, google-cloud-compute) while Azure splits management-plane from data-plane packages (azure-mgmt-storage versus azure-storage-blob) — picking the data-plane package when you needed the management plane is the usual first mistake.

Does a boto3 call inside a Pulumi program run during preview?

Yes, unless you stop it. Python module-level and constructor code executes on every pulumi preview as well as every pulumi up, so a mutating call placed there fires during what you thought was a dry run. Guard imperative work with pulumi.runtime.is_dry_run() and keep it out of component constructors entirely.

How do I stop an SDK-created resource from being orphaned?

Tag it at creation with a stable logical name and make the create path a filtered describe first, so a re-run adopts the existing resource instead of making a second one. For anything with a real lifecycle, model it in the framework — pulumi import or a CDKTF import block will bring an existing resource under management without recreating it.

Should the SDK layer live in the same repository as the Pulumi program?

Yes, as a separate package with its own tests. Shared typing and one credential factory are worth more than repository separation, and a same-repo module means a change to the lookup surface shows up in the same pull request as the infrastructure change that motivated it.

Related

• Best practices for managing cloud credentials in Python — typed credential loaders, secret handling, and rotation-safe state recovery for every provider chain.
• Using boto3 inside Pulumi and CDKTF — when to drop to the AWS SDK for lookups and imperative steps without corrupting framework state.
• Python IaC Fundamentals & Strategy — the parent overview tying SDK integration to design principles, tooling choice, and security.