Domain 1: Development with AWS Services (32%)¶

# Use IAM roles (preferred for EC2/Lambda)
import boto3
client = boto3.client('s3')

# Explicit credentials (avoid hardcoding)
client = boto3.client(
    's3',
    aws_access_key_id='ACCESS_KEY',
    aws_secret_access_key='SECRET_KEY',
    region_name='us-east-1'
)

# Use environment variables or credential files

import boto3
from botocore.exceptions import ClientError
from botocore.config import Config

# Configure exponential backoff
config = Config(
    retries={
        'max_attempts': 3,
        'mode': 'adaptive'
    }
)

client = boto3.client('dynamodb', config=config)

try:
    response = client.put_item(
        TableName='MyTable',
        Item={'id': {'S': '123'}}
    )
except ClientError as e:
    if e.response['Error']['Code'] == 'ResourceNotFoundException':
        print("Table not found")
    elif e.response['Error']['Code'] == 'ProvisionedThroughputExceededException':
        print("Throttled - retry with backoff")
    else:
        raise

import boto3
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('Users')

# Put item
table.put_item(Item={'userId': '123', 'name': 'John', 'email': 'john@example.com'})

# Get item
response = table.get_item(Key={'userId': '123'})
item = response['Item']

# Query (requires partition key)
response = table.query(
    KeyConditionExpression='userId = :uid',
    ExpressionAttributeValues={':uid': '123'}
)

# Scan (full table scan - avoid in production)
response = table.scan(
    FilterExpression='age > :age',
    ExpressionAttributeValues={':age': 25}
)

# Batch operations
with table.batch_writer() as batch:
    for i in range(100):
        batch.put_item(Item={'userId': str(i), 'name': f'User{i}'})

# Transactions
client = boto3.client('dynamodb')
client.transact_write_items(
    TransactItems=[
        {'Put': {'TableName': 'Users', 'Item': {'userId': {'S': '123'}}}},
        {'Update': {'TableName': 'Orders', 'Key': {'orderId': {'S': '456'}}}},
    ]
)

# Enable streams on table
client.update_table(
    TableName='MyTable',
    StreamSpecification={
        'StreamEnabled': True,
        'StreamViewType': 'NEW_AND_OLD_IMAGES'
    }
)

# Lambda trigger for DynamoDB Streams
def lambda_handler(event, context):
    for record in event['Records']:
        if record['eventName'] == 'INSERT':
            new_image = record['dynamodb']['NewImage']
            # Process new item
        elif record['eventName'] == 'MODIFY':
            old_image = record['dynamodb']['OldImage']
            new_image = record['dynamodb']['NewImage']
            # Process update

import boto3
s3 = boto3.client('s3')

# Upload file
s3.upload_file('local-file.txt', 'my-bucket', 'remote-file.txt')

# Download file
s3.download_file('my-bucket', 'remote-file.txt', 'downloaded-file.txt')

# Put object with metadata
s3.put_object(
    Bucket='my-bucket',
    Key='data.json',
    Body=json.dumps(data),
    ContentType='application/json',
    Metadata={'user': 'john', 'version': '1.0'}
)

# Generate presigned URL
url = s3.generate_presigned_url(
    'get_object',
    Params={'Bucket': 'my-bucket', 'Key': 'file.txt'},
    ExpiresIn=3600  # 1 hour
)

# Multipart upload for large files
mpu = s3.create_multipart_upload(Bucket='my-bucket', Key='large-file.zip')
upload_id = mpu['UploadId']

import pymysql
import os

# Connection pooling for Lambda
connection = None

def get_connection():
    global connection
    if connection is None or not connection.open:
        connection = pymysql.connect(
            host=os.environ['DB_HOST'],
            user=os.environ['DB_USER'],
            password=os.environ['DB_PASSWORD'],
            database=os.environ['DB_NAME'],
            connect_timeout=5
        )
    return connection

def lambda_handler(event, context):
    conn = get_connection()
    with conn.cursor() as cursor:
        cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
        result = cursor.fetchone()
    return result

import boto3
sqs = boto3.client('sqs')

# Send message
response = sqs.send_message(
    QueueUrl='https://sqs.us-east-1.amazonaws.com/123456789012/MyQueue',
    MessageBody='Hello World',
    MessageAttributes={
        'Priority': {'StringValue': 'High', 'DataType': 'String'}
    }
)

# Receive messages
messages = sqs.receive_message(
    QueueUrl=queue_url,
    MaxNumberOfMessages=10,
    WaitTimeSeconds=20,  # Long polling
    VisibilityTimeout=30
)

# Delete message after processing
sqs.delete_message(
    QueueUrl=queue_url,
    ReceiptHandle=message['ReceiptHandle']
)

# Send batch messages
entries = [
    {'Id': '1', 'MessageBody': 'Message 1'},
    {'Id': '2', 'MessageBody': 'Message 2'}
]
sqs.send_message_batch(QueueUrl=queue_url, Entries=entries)

import boto3
sns = boto3.client('sns')

# Publish to topic
response = sns.publish(
    TopicArn='arn:aws:sns:us-east-1:123456789012:MyTopic',
    Message='Hello World',
    Subject='Notification',
    MessageAttributes={
        'type': {'DataType': 'String', 'StringValue': 'alert'}
    }
)

# Fan-out pattern: SNS → multiple SQS queues
# Subscribe SQS queues to SNS topic for parallel processing