advancedgeneralAmazon styleStripe styleCapital One styleDatadog style

Amazon platform interview: Write a Python boto3 inventory collector for 200 AWS accounts. How do you handle pagination, STS role assumption, retries, throttling, and partial failure reporting?

⚡

Quick Answer

Use STS AssumeRole per account, create regional clients from temporary credentials, consume paginator APIs instead of single calls, configure botocore retries, and process accounts with bounded concurrency. Record per-account success, throttles, access-denied errors, and skipped regions so the inventory is trustworthy even when some accounts fail.

Detailed Answer

Imagine asking 200 branch offices to send you a list of every laptop they own. If you call only the first office, stop after the first page, or hide which offices did not answer, the final spreadsheet looks clean but lies. AWS inventory automation has the same failure mode: the script can produce a file while silently missing accounts, regions, or pages.

The core boto3 pattern is explicit. First call STS AssumeRole into the target account using a role created for inventory. Then build service clients with those temporary credentials and the target region. For list-style APIs, use paginators because many AWS APIs return only a limited page and a continuation token. The boto3 retry guide documents standard and adaptive retry modes, and production collectors should set retry mode and max attempts intentionally rather than relying on unknown environment defaults.

Concurrency must be bounded. A script that fans out 200 accounts times 25 regions times 10 services can create tens of thousands of API calls. Unbounded ThreadPoolExecutor usage will cause throttling, noisy CloudTrail, and sometimes organization-wide guardrail alerts. A mature collector uses a small worker pool, per-service backoff, and clear retry classification: throttling and transient network errors are retried, AccessDenied is recorded as an authorization gap, and validation errors are treated as code or configuration defects.

The output format matters as much as the API calls. Each account-region-service tuple should report status, item count, elapsed time, and error class. That lets platform teams distinguish no resources from no permission. For auditability, include the assumed role ARN, collection timestamp, script version, and schema version in the artifact. Large companies ask this because inventory data drives security posture, cost cleanup, and incident response.

The non-obvious gotcha is eventual consistency. A newly created role, instance, tag, or region opt-in may not appear immediately. Senior candidates mention rerunning collectors, comparing deltas, and designing consumers to tolerate late-arriving data instead of treating one scan as absolute truth.

from __future__ import annotations # Use modern type annotations. from concurrent.futures import ThreadPoolExecutor, as_completed # Bound account fanout. import boto3 # AWS SDK for Python. from botocore.config import Config # Configure retries and timeouts explicitly. from botocore.exceptions import ClientError # Classify AWS API failures. AWS_CONFIG = Config( # Shared botocore behavior for all clients. retries={"mode": "standard", "max_attempts": 6}, # Retry transient AWS errors predictably. connect_timeout=3, # Fail fast when network paths are broken. read_timeout=20, # Bound slow API responses. ) # Config is passed into each boto3 client. def assume_inventory_role(account_id: str) -> dict: # Get temporary account credentials. sts = boto3.client("sts", config=AWS_CONFIG) # Create an STS client in the caller account. resp = sts.assume_role( # Ask AWS for temporary credentials in the target account. RoleArn=f"arn:aws:iam::{account_id}:role/OrganizationInventoryReadOnly", # Target role. RoleSessionName="pydevops-inventory", # Session name visible in CloudTrail. ) # STS returns time-limited credentials. return resp["Credentials"] # Return credentials for regional clients. def ec2_client(creds: dict, region: str): # Build a regional EC2 client. return boto3.client( # Construct client with assumed-role credentials. "ec2", # Inventory EC2 instances in this example. region_name=region, # Query one region at a time. aws_access_key_id=creds["AccessKeyId"], # Temporary access key. aws_secret_access_key=creds["SecretAccessKey"], # Temporary secret key. aws_session_token=creds["SessionToken"], # Temporary session token. config=AWS_CONFIG, # Apply retry and timeout settings. ) # Return configured client. def collect_account_region(account_id: str, region: str) -> dict: # Collect one unit of work. try: # Keep failures scoped to this account-region pair. creds = assume_inventory_role(account_id) # Assume the inventory role. ec2 = ec2_client(creds, region) # Create EC2 client for the region. paginator = ec2.get_paginator("describe_instances") # Use paginator, not one API call. instances = [] # Store normalized inventory records. for page in paginator.paginate(PaginationConfig={"PageSize": 100}): # Walk every page. for reservation in page.get("Reservations", []): # Reservations contain instances. for inst in reservation.get("Instances", []): # Normalize each instance. instances.append({"account": account_id, "region": region, "id": inst["InstanceId"], "state": inst["State"]["Name"]}) # Keep essential fields. return {"account": account_id, "region": region, "status": "ok", "count": len(instances), "items": instances} # Report success. except ClientError as exc: # AWS returned a structured error. code = exc.response.get("Error", {}).get("Code", "ClientError") # Extract error code. return {"account": account_id, "region": region, "status": "error", "error": code, "items": []} # Preserve partial failure. def collect_all(accounts: list[str], regions: list[str]) -> list[dict]: # Run bounded fanout. jobs = [(acct, region) for acct in accounts for region in regions] # Build explicit work list. results = [] # Accumulate every account-region result. with ThreadPoolExecutor(max_workers=16) as pool: # Bound API pressure. futures = [pool.submit(collect_account_region, acct, region) for acct, region in jobs] # Submit jobs. for future in as_completed(futures): # Consume results as they complete. results.append(future.result()) # Keep successes and recorded failures. return results # Caller can write JSON, metrics, or reports.

Amazon platform interview: Write a Python boto3 inventory collector for 200 AWS accounts. How do you handle pagination, STS role assumption, retries, throttling, and partial failure reporting?

More Python DevOps Automation questions