aws

Use when writing boto3 or AWS SDK v3 code — configuring IAM auth, reading/writing S3, designing DynamoDB access patterns, writing Lambda handlers, processing SQS batches, or troubleshooting credential

SKILL.md

GitHub

AWS SDK Patterns

Production patterns for AWS services using boto3 (Python) and AWS SDK v3 (TypeScript).

When to Activate

  • Writing code that imports boto3, @aws-sdk/*, or aws-sdk
  • Configuring authentication for AWS services (IAM roles, instance profiles, environment credentials)
  • Reading or writing objects in S3 with presigned URLs or multipart upload
  • Designing a DynamoDB table or writing query/scan access patterns
  • Writing Lambda handler functions or connecting them to SQS/S3/API Gateway triggers
  • Processing SQS messages with partial batch failure handling
  • Retrieving secrets from Secrets Manager or Parameter Store
  • Troubleshooting ClientError, credential resolution, or throttling

Authentication

Credential Chain (always prefer role-based auth)

boto3 and AWS SDK v3 resolve credentials in this order — the same code works locally and in production without changes:

1. Explicit credentials passed to client (avoid — hardcodes secrets)
2. Environment variables: AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_SESSION_TOKEN
3. AWS config file: ~/.aws/credentials
4. IAM instance profile (EC2) / task role (ECS) / execution role (Lambda)
5. IAM Roles Anywhere / container credentials

# Python — boto3 picks up credentials automatically
import boto3

s3 = boto3.client("s3", region_name="us-east-1")  # uses credential chain

# BAD: hardcoded credentials
s3 = boto3.client("s3", aws_access_key_id="AKIA...", aws_secret_access_key="...")

# GOOD: explicit profile for local dev only
session = boto3.Session(profile_name="dev")
s3 = session.client("s3")

// TypeScript — SDK v3 uses same chain automatically
import { S3Client } from "@aws-sdk/client-s3";

const s3 = new S3Client({ region: "us-east-1" });  // no credentials needed

// Local dev with named profile
import { fromIni } from "@aws-sdk/credential-providers";
const s3 = new S3Client({
  region: "us-east-1",
  credentials: fromIni({ profile: "dev" }),
});

IAM Role Decision Matrix

Environment Auth method How to set up
Local dev Named profile (~/.aws/credentials) aws configure --profile dev
GitHub Actions OIDC + IAM role (no long-lived keys) aws-actions/configure-aws-credentials with role-to-assume
Lambda Execution role (auto-injected) Attach IAM role to function in console/IaC
ECS / Fargate Task role taskRoleArn in task definition
EC2 Instance profile Attach IAM role to instance
Local → assume role AWS_PROFILE + role ARN Add role_arn to ~/.aws/config

Configuration

from pydantic_settings import BaseSettings

class AWSSettings(BaseSettings):
    aws_region: str = "us-east-1"
    s3_bucket: str
    dynamodb_table: str
    sqs_queue_url: str
    secrets_manager_prefix: str = "/myapp/prod"

    class Config:
        env_file = ".env"

settings = AWSSettings()

Never store AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY in .env files committed to source control.

S3

Upload and Download

import boto3
from botocore.exceptions import ClientError

s3 = boto3.client("s3", region_name="us-east-1")

def upload_object(bucket: str, key: str, data: bytes, content_type: str = "application/octet-stream") -> str:
    s3.put_object(Bucket=bucket, Key=key, Body=data, ContentType=content_type)
    return f"s3://{bucket}/{key}"

def download_object(bucket: str, key: str) -> bytes:
    response = s3.get_object(Bucket=bucket, Key=key)
    return response["Body"].read()

def list_objects(bucket: str, prefix: str = "") -> list[str]:
    paginator = s3.get_paginator("list_objects_v2")
    keys = []
    for page in paginator.paginate(Bucket=bucket, Prefix=prefix):
        keys.extend(obj["Key"] for obj in page.get("Contents", []))
    return keys

import { S3Client, PutObjectCommand, GetObjectCommand } from "@aws-sdk/client-s3";
import { Readable } from "stream";

const s3 = new S3Client({ region: "us-east-1" });

async function uploadObject(bucket: string, key: string, body: Buffer, contentType: string) {
  await s3.send(new PutObjectCommand({ Bucket: bucket, Key: key, Body: body, ContentType: contentType }));
  return `s3://${bucket}/${key}`;
}

async function downloadObject(bucket: string, key: string): Promise<Buffer> {
  const res = await s3.send(new GetObjectCommand({ Bucket: bucket, Key: key }));
  return Buffer.from(await res.Body!.transformToByteArray());
}

Presigned URLs

from datetime import timedelta

def get_presigned_url(bucket: str, key: str, expiry_seconds: int = 3600) -> str:
    return s3.generate_presigned_url(
        "get_object",
        Params={"Bucket": bucket, "Key": key},
        ExpiresIn=expiry_seconds,
    )

def get_presigned_upload_url(bucket: str, key: str, content_type: str, expiry_seconds: int = 900) -> str:
    return s3.generate_presigned_url(
        "put_object",
        Params={"Bucket": bucket, "Key": key, "ContentType": content_type},
        ExpiresIn=expiry_seconds,
    )

Multipart Upload (files > 100 MB)

import boto3
from boto3.s3.transfer import TransferConfig

s3_resource = boto3.resource("s3")

config = TransferConfig(
    multipart_threshold=100 * 1024 * 1024,   # 100 MB
    multipart_chunksize=50 * 1024 * 1024,    # 50 MB chunks
    max_concurrency=10,
)

def upload_large_file(bucket: str, key: str, file_path: str):
    s3_resource.Object(bucket, key).upload_file(file_path, Config=config)

DynamoDB

Single-Table Design

Table: MyApp
PK           SK                     Attributes
USER#u1      PROFILE                name, email, plan
USER#u1      ORDER#2024-01          status, total
USER#u1      ORDER#2024-02          status, total
ORDER#o1     METADATA               customer_id, created_at
ORDER#o1     ITEM#sku-a             qty, unit_price

GSI1: GSI1PK=customer_id, GSI1SK=created_at → query all orders for a customer

Access patterns map to key structure — define all access patterns before writing schema.

CRUD Operations

import boto3
from boto3.dynamodb.conditions import Key, Attr
from decimal import Decimal

dynamodb = boto3.resource("dynamodb", region_name="us-east-1")
table = dynamodb.Table("MyApp")

def put_item(item: dict) -> None:
    table.put_item(Item=item)

def get_item(pk: str, sk: str) -> dict | None:
    response = table.get_item(Key={"PK": pk, "SK": sk})
    return response.get("Item")

def query_items(pk: str, sk_prefix: str) -> list[dict]:
    response = table.query(
        KeyConditionExpression=Key("PK").eq(pk) & Key("SK").begins_with(sk_prefix),
    )
    return response["Items"]

def update_item(pk: str, sk: str, updates: dict) -> None:
    expr = "SET " + ", ".join(f"#{k} = :{k}" for k in updates)
    table.update_item(
        Key={"PK": pk, "SK": sk},
        UpdateExpression=expr,
        ExpressionAttributeNames={f"#{k}": k for k in updates},
        ExpressionAttributeValues={f":{k}": v for k, v in updates.items()},
    )

def delete_item(pk: str, sk: str) -> None:
    table.delete_item(Key={"PK": pk, "SK": sk})

Batch Operations

def batch_write(items: list[dict]) -> None:
    with table.batch_writer() as batch:  # auto-handles 25-item limit and unprocessed items
        for item in items:
            batch.put_item(Item=item)

def batch_get(keys: list[dict]) -> list[dict]:
    response = dynamodb.meta.client.batch_get_item(
        RequestItems={table.name: {"Keys": keys[:100]}}  # max 100 per call
    )
    return response["Responses"].get(table.name, [])

DynamoDB vs RDS Decision

Factor DynamoDB RDS (Postgres)
Access patterns Known, finite, key-based Ad-hoc queries, complex joins
Scale Millions of req/s, auto-scale Vertical + read replicas
Schema Flexible, item-level Strict, table-level
Consistency Eventually consistent (default) ACID
Operational cost ~Zero ops Patching, backups, failover
Cost model Pay-per-request or provisioned Instance hours

Lambda

Handler Patterns

# Python — structured handler with typed events
import json, logging
from typing import Any

logger = logging.getLogger()
logger.setLevel(logging.INFO)

def handler(event: dict, context: Any) -> dict:
    logger.info("invocation", extra={"request_id": context.aws_request_id, "event": event})

    try:
        result = process(event)
        return {"statusCode": 200, "body": json.dumps(result)}
    except ValueError as e:
        return {"statusCode": 400, "body": json.dumps({"error": str(e)})}
    except Exception:
        logger.exception("unhandled_error")
        raise  # let Lambda retry / send to DLQ

def process(event: dict) -> dict:
    ...

// TypeScript — API Gateway proxy event
import { APIGatewayProxyEvent, APIGatewayProxyResult, Context } from "aws-lambda";

export const handler = async (
  event: APIGatewayProxyEvent,
  context: Context,
): Promise<APIGatewayProxyResult> => {
  const body = JSON.parse(event.body ?? "{}");
  try {
    const result = await process(body);
    return { statusCode: 200, body: JSON.stringify(result) };
  } catch (err) {
    console.error({ requestId: context.awsRequestId, err });
    return { statusCode: 500, body: JSON.stringify({ error: "internal error" }) };
  }
};

Lambda Environment Best Practices

import os
import boto3

# Initialize clients outside the handler — reused across warm invocations
_s3 = None
_table = None

def get_s3():
    global _s3
    if _s3 is None:
        _s3 = boto3.client("s3")
    return _s3

def get_table():
    global _table
    if _table is None:
        dynamodb = boto3.resource("dynamodb")
        _table = dynamodb.Table(os.environ["DYNAMODB_TABLE"])
    return _table

# Read config from environment, not hardcoded
BUCKET = os.environ["S3_BUCKET"]
REGION = os.environ.get("AWS_REGION", "us-east-1")

Lambda Event Source Comparison

Trigger Invocation Retry behavior Batch
API Gateway / ALB Sync None (caller handles) No
SQS Async Redrive to DLQ after maxReceiveCount Yes (up to 10000)
S3 Async 2 retries then discard No (one event per object)
DynamoDB Streams Async Retry until success or record expires Yes (per shard)
EventBridge Async Configurable retry + DLQ No
SNS Async 3 retries then DLQ No

SQS

Send and Receive

sqs = boto3.client("sqs", region_name="us-east-1")
QUEUE_URL = os.environ["SQS_QUEUE_URL"]

def send_message(body: dict, deduplication_id: str | None = None) -> str:
    params = {"QueueUrl": QUEUE_URL, "MessageBody": json.dumps(body)}
    if deduplication_id:  # required for FIFO queues
        params["MessageDeduplicationId"] = deduplication_id
        params["MessageGroupId"] = body.get("group_id", "default")
    response = sqs.send_message(**params)
    return response["MessageId"]

def send_batch(messages: list[dict]) -> None:
    entries = [
        {"Id": str(i), "MessageBody": json.dumps(msg)}
        for i, msg in enumerate(messages[:10])  # max 10 per batch
    ]
    response = sqs.send_message_batch(QueueUrl=QUEUE_URL, Entries=entries)
    if response.get("Failed"):
        raise RuntimeError(f"batch send failed: {response['Failed']}")

Lambda SQS Consumer with Partial Batch Failure

def handler(event: dict, context: Any) -> dict:
    batch_item_failures = []

    for record in event["Records"]:
        message_id = record["messageId"]
        try:
            body = json.loads(record["body"])
            process_message(body)
        except Exception:
            logger.exception("message_failed", extra={"message_id": message_id})
            batch_item_failures.append({"itemIdentifier": message_id})

    # Return only failed IDs — SQS retries these, deletes the rest
    return {"batchItemFailures": batch_item_failures}

Enable partial batch failure (FunctionResponseTypes: [ReportBatchItemFailures]) in the event source mapping — otherwise one failure requeues the entire batch.

Secrets Manager

import json
import boto3
from functools import lru_cache

_sm = boto3.client("secretsmanager", region_name="us-east-1")

@lru_cache(maxsize=None)  # cache per Lambda warm instance
def get_secret(secret_name: str) -> dict:
    response = _sm.get_secret_value(SecretId=secret_name)
    raw = response.get("SecretString") or response["SecretBinary"].decode()
    try:
        return json.loads(raw)
    except json.JSONDecodeError:
        return {"value": raw}

# Usage
db_creds = get_secret("/myapp/prod/db")
password = db_creds["password"]

Secrets Manager vs Parameter Store

Factor Secrets Manager Parameter Store (SSM)
Secret rotation Built-in (Lambda-based) Manual
Versioning Yes Yes
Cost $0.40/secret/month Free (standard), $0.05/10K API calls advanced
Size limit 64 KB 4 KB (standard), 8 KB (advanced)
Best for DB passwords, API keys, rotation Config values, feature flags, non-sensitive config 
# Parameter Store (cheaper for non-secrets)
ssm = boto3.client("ssm")

def get_parameter(name: str, with_decryption: bool = True) -> str:
    response = ssm.get_parameter(Name=name, WithDecryption=with_decryption)
    return response["Parameter"]["Value"]

Retry and Error Handling

botocore Retry Config

from botocore.config import Config

retry_config = Config(
    retries={
        "max_attempts": 5,
        "mode": "adaptive",   # adaptive > standard > legacy; backs off on throttle
    },
    connect_timeout=5,
    read_timeout=30,
)

s3 = boto3.client("s3", config=retry_config)
dynamodb = boto3.client("dynamodb", config=retry_config)

ClientError Handling

from botocore.exceptions import ClientError, NoCredentialsError

def safe_get_object(bucket: str, key: str) -> bytes | None:
    try:
        return s3.get_object(Bucket=bucket, Key=key)["Body"].read()
    except ClientError as e:
        code = e.response["Error"]["Code"]
        match code:
            case "NoSuchKey" | "404":
                return None
            case "AccessDenied" | "403":
                logger.error("s3_access_denied", bucket=bucket, key=key)
                raise
            case "ThrottlingException" | "RequestLimitExceeded" | "SlowDown":
                raise  # botocore retry handles this
            case _:
                logger.error("s3_error", code=code, bucket=bucket, key=key)
                raise
    except NoCredentialsError:
        logger.critical("no_aws_credentials")
        raise

Common Error Codes

Service Error Code Meaning
S3 NoSuchKey Object doesn't exist
S3 NoSuchBucket Bucket doesn't exist or no access
DynamoDB ConditionalCheckFailedException Optimistic lock / condition failed
DynamoDB ProvisionedThroughputExceededException Throttled — retry with backoff
DynamoDB ResourceNotFoundException Table doesn't exist
Secrets Manager ResourceNotFoundException Secret not found
All AccessDeniedException IAM permissions missing
All ThrottlingException Rate limited — botocore retries

IAM Least Privilege

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["s3:GetObject", "s3:PutObject"],
      "Resource": "arn:aws:s3:::my-bucket/uploads/*"
    },
    {
      "Effect": "Allow",
      "Action": ["dynamodb:GetItem", "dynamodb:PutItem", "dynamodb:Query", "dynamodb:UpdateItem"],
      "Resource": [
        "arn:aws:dynamodb:us-east-1:123456789012:table/MyApp",
        "arn:aws:dynamodb:us-east-1:123456789012:table/MyApp/index/*"
      ]
    },
    {
      "Effect": "Allow",
      "Action": "secretsmanager:GetSecretValue",
      "Resource": "arn:aws:secretsmanager:us-east-1:123456789012:secret:/myapp/prod/*"
    }
  ]
}

# BAD: wildcard on resource
"Action": "s3:*", "Resource": "*"

# BAD: admin permissions for app role
"Action": "*", "Resource": "*"

# GOOD: specific actions, specific ARNs with path constraints

Cost Controls

Lever Impact How
DynamoDB on-demand vs provisioned High On-demand for unpredictable traffic; provisioned + auto-scaling for steady workloads
S3 storage classes Medium Lifecycle policy: Standard → Standard-IA after 30d → Glacier after 90d
Lambda memory sizing Medium Profile with Lambda Power Tuning; more memory often runs faster and costs less
DynamoDB DAX cache Medium Cache read-heavy tables; reduces read capacity units
S3 request costs Low-medium Use CloudFront in front of S3 for high-volume GET patterns
Secrets Manager calls Low Cache secrets in Lambda warm instance; don't call on every invocation
CloudWatch Logs retention Low Set retention (7–30d) — default is forever

See also: event-driven, caching, observability

Red Flags

  • Hardcoded aws_access_key_id in code or config files — long-term credentials in code are a top AWS compromise vector; use the credential chain (IAM role, instance profile, environment variable)
  • * in IAM policy actions or resources — wildcard policies grant far more than needed; scope every policy to the minimum set of actions and specific resource ARNs
  • DynamoDB Scan in production code paths — Scan reads every item in the table and consumes all provisioned capacity; design access patterns around Query using primary keys and GSIs
  • Lambda handler that creates DB or SDK connections on every invocation — connections initialized inside the handler are destroyed and recreated per call; initialize SDK clients outside the handler in module scope
  • SQS visibility timeout shorter than Lambda timeout — if visibility timeout < Lambda timeout, the message becomes visible before processing finishes, causing duplicate delivery; set visibility timeout to 6× Lambda timeout
  • S3 presigned URLs without a short expiry — presigned URLs with a far-future expiry can be bookmarked and reused long after the intended access window; always set the shortest practical expiry
  • CloudWatch logs without structured JSON — unstructured log lines can't be queried with CloudWatch Insights; emit JSON with consistent fields (level, message, correlation_id) from every Lambda

Checklist

  • All SDK clients use the credential chain — no hardcoded aws_access_key_id or aws_secret_access_key
  • Lambda, ECS tasks, and EC2 instances use IAM roles/profiles — no long-lived keys in env vars
  • IAM policies follow least privilege: specific actions, specific resource ARNs
  • GitHub Actions uses OIDC role assumption — no IAM user keys stored as secrets
  • boto3/SDK clients initialized outside Lambda handler (reused on warm invocations)
  • botocore retry config set to adaptive mode with max_attempts=5
  • ClientError caught and branched by error code — not caught and swallowed
  • SQS Lambda consumer returns batchItemFailures for partial batch failure
  • DynamoDB access patterns defined before schema — table designed around queries
  • S3 lifecycle policy configured for infrequently accessed data
  • Secrets retrieved from Secrets Manager or Parameter Store — not environment variables
  • CloudWatch Logs retention period set on all log groups

Похожие навыки: other

limits-colimits

Context management for Claude Code. Hooks maintain state via ledgers and handoffs. MCP execution without context polluti...

Автор: parcadei
3,392

ddttom-allaboutv2-claude-skills-block-collection-and-party

The Block Collection and Block Party are repositories for existing AEM blocks, build tools, code snippets, integration p...

Автор: mattnigh
20

nodnarbnitram-claude-code-extensions-claude-skills-esphome-confi

This skill should be used when the user asks to "create an esphome config", "set up an esp32 device", "configure esphome...

Автор: mattnigh
20