1
# Synchronous invoke via CLI
2
aws lambda invoke \
3
  --function-name myFunction \
4
  --payload '{"key":"value"}' \
5
  response.json

1
import os
2
DB_HOST = os.environ['DB_HOST']
3
API_KEY = os.environ['API_KEY']

1
def handler(event, context):
2
    print(context.function_name)
3
    print(context.get_remaining_time_in_millis())
4
    return {"statusCode": 200, "body": "OK"}

1
# Lambda ARN using stage variable
2
arn:aws:lambda:us-east-1:123:function:myFunc:${stageVariables.lambdaAlias}

1
import boto3
2
s3 = boto3.client('s3')
3
url = s3.generate_presigned_url(
4
    'get_object',
5
    Params={'Bucket': 'mybucket', 'Key': 'file.txt'},
6
    ExpiresIn=3600
7
)

1
# boto3 automatically uses the Lambda execution role
2
import boto3
3
s3 = boto3.client('s3')
4
s3.list_buckets()

1
sts = boto3.client('sts')
2
response = sts.assume_role(
3
    RoleArn='arn:aws:iam::123456789:role/MyRole',
4
    RoleSessionName='MySession'
5
)
6
creds = response['Credentials']

1
kms = boto3.client('kms')
2
response = kms.generate_data_key(
3
    KeyId='alias/mykey',
4
    KeySpec='AES_256'
5
)
6
plaintext_key = response['Plaintext']
7
encrypted_key = response['CiphertextBlob']

1
client = boto3.client('secretsmanager')
2
secret = client.get_secret_value(SecretId='prod/db/password')
3
password = secret['SecretString']

1
ssm = boto3.client('ssm')
2
param = ssm.get_parameter(
3
    Name='/myapp/db/password',
4
    WithDecryption=True
5
)
6
value = param['Parameter']['Value']

1
# S3 CORS config
2
[{"AllowedOrigins": ["https://myapp.com"],
3
  "AllowedMethods": ["GET","PUT"],
4
  "AllowedHeaders": ["*"]}]

1
version: 0.0
2
os: linux
3
files:
4
  - source: /
5
    destination: /var/www/html
6
hooks:
7
  AfterInstall:
8
    - location: scripts/install_deps.sh
9
  ApplicationStart:
10
    - location: scripts/start_server.sh

1
# Stack A export
2
Outputs:
3
  VpcId:
4
    Value: !Ref MyVPC
5
    Export:
6
      Name: MyVPC-ID
7

8
# Stack B import
9
Resources:
10
  MyInstance:
11
    Properties:
12
      VpcId: !ImportValue MyVPC-ID

1
# SAM canary deployment
2
AutoPublishAlias: live
3
DeploymentPreference:
4
  Type: Canary10Percent10Minutes

1
# Python instrument boto3 with X-Ray
2
from aws_xray_sdk.core import xray_recorder, patch_all
3
patch_all()  # patches boto3, requests, etc.

1
import time, random
2
for attempt in range(5):
3
    try:
4
        response = call_aws_api()
5
        break
6
    except ThrottlingException:
7
        time.sleep((2 ** attempt) + random.random())

1
# Find Lambda errors in the last hour
2
fields @timestamp, @message
3
| filter @message like /ERROR/
4
| sort @timestamp desc
5
| limit 50

1
$ aws configure
2
AWS Access Key ID: AKIA...
3
AWS Secret Access Key: ****
4
Default region name: us-east-1
5
Default output format: json

1
$ aws configure --profile dev
2
$ aws s3 ls --profile dev
3
$ export AWS_PROFILE=dev

1
$ aws ec2 describe-instances \
2
    --query 'Reservations[*].Instances[*].InstanceId' \
3
    --output text

1
import boto3
2
from botocore.config import Config
3

4
config = Config(
5
    retries={'max_attempts': 10, 'mode': 'standard'}
6
)
7
client = boto3.client('dynamodb', config=config)

1
paginator = boto3.client('s3').get_paginator('list_objects_v2')
2
for page in paginator.paginate(Bucket='mybucket'):
3
    for obj in page['Contents']:
4
        print(obj['Key'])

1
ec2 = boto3.client('ec2')
2
waiter = ec2.get_waiter('instance_running')
3
waiter.wait(InstanceIds=['i-1234567890abcdef0'])
4
print("Instance is running!")

1
# LocalStack local testing
2
s3 = boto3.client(
3
    's3',
4
    region_name='us-east-1',
5
    endpoint_url='http://localhost:4566'
6
)

1
import { DynamoDBClient, GetItemCommand } from "@aws-sdk/client-dynamodb";
2
const client = new DynamoDBClient({ region: "us-east-1" });
3
const response = await client.send(
4
  new GetItemCommand({ TableName: "Users", Key: { id: { S: "123" } } })
5
);

1
// SDK v3 with lib-dynamodb
2
import { DynamoDBDocumentClient, GetCommand } from "@aws-sdk/lib-dynamodb";
3
const docClient = DynamoDBDocumentClient.from(client);
4
const result = await docClient.send(
5
  new GetCommand({ TableName: "Users", Key: { id: "123" } })
6
);

1
try:
2
    table.get_item(Key={'id': '123'})
3
except ClientError as e:
4
    code = e.response['Error']['Code']
5
    if code == 'ProvisionedThroughputExceededException':
6
        time.sleep(2)

1
$ aws cloudformation deploy \
2
    --template-file template.yaml \
3
    --stack-name my-app \
4
    --capabilities CAPABILITY_IAM \
5
    --parameter-overrides Env=prod

1
from aws_cdk import Stack, aws_lambda as _lambda
2
class MyStack(Stack):
3
    def __init__(self, scope, id, **kwargs):
4
        super().__init__(scope, id, **kwargs)
5
        _lambda.Function(self, "MyFunc",
6
            runtime=_lambda.Runtime.PYTHON_3_11,
7
            handler="app.handler",
8
            code=_lambda.Code.from_asset("lambda"))

1
# boto3 handles multipart automatically with transfer config
2
from boto3.s3.transfer import TransferConfig
3
config = TransferConfig(multipart_threshold=1024*25)
4
s3.upload_file('largefile.zip', 'mybucket', 'largefile.zip',
5
               Config=config)

1
table.put_item(
2
    Item={'id': '123', 'version': 1, 'name': 'Ali'},
3
    ConditionExpression='attribute_not_exists(id)'
4
)
5
# Fails if item with id=123 already exists