AWS SDK Integration

Node.js (SDK v3)

Client Setup

import { DynamoDBClient } from '@aws-sdk/client-dynamodb';
import { DynamoDBDocumentClient, PutCommand, GetCommand, QueryCommand } from '@aws-sdk/lib-dynamodb';

const client = new DynamoDBClient({ region: process.env.AWS_REGION });
const ddb = DynamoDBDocumentClient.from(client);

DynamoDB

// Put item
await ddb.send(new PutCommand({
  TableName: 'Users',
  Item: { pk: `USER#${id}`, sk: 'PROFILE', name, email, createdAt: Date.now() },
  ConditionExpression: 'attribute_not_exists(pk)', // Prevent overwrite
}));

// Get item
const { Item } = await ddb.send(new GetCommand({
  TableName: 'Users', Key: { pk: `USER#${id}`, sk: 'PROFILE' },
}));

// Query
const { Items } = await ddb.send(new QueryCommand({
  TableName: 'Orders',
  KeyConditionExpression: 'pk = :pk AND begins_with(sk, :prefix)',
  ExpressionAttributeValues: { ':pk': `USER#${userId}`, ':prefix': 'ORDER#' },
}));

Lambda Invocation

import { LambdaClient, InvokeCommand } from '@aws-sdk/client-lambda';

const lambda = new LambdaClient({});
const { Payload } = await lambda.send(new InvokeCommand({
  FunctionName: 'processOrder',
  InvocationType: 'Event', // async ('RequestResponse' for sync)
  Payload: JSON.stringify({ orderId }),
}));

SNS

import { SNSClient, PublishCommand } from '@aws-sdk/client-sns';

const sns = new SNSClient({});
await sns.send(new PublishCommand({
  TopicArn: process.env.ORDER_TOPIC_ARN,
  Message: JSON.stringify({ orderId, status: 'completed' }),
  MessageAttributes: {
    eventType: { DataType: 'String', StringValue: 'ORDER_COMPLETED' },
  },
}));

CloudWatch Metrics

import { CloudWatchClient, PutMetricDataCommand } from '@aws-sdk/client-cloudwatch';

const cw = new CloudWatchClient({});
await cw.send(new PutMetricDataCommand({
  Namespace: 'MyApp',
  MetricData: [{
    MetricName: 'OrderProcessingTime',
    Value: durationMs,
    Unit: 'Milliseconds',
    Dimensions: [{ Name: 'Service', Value: 'OrderProcessor' }],
  }],
}));

Python (boto3)

import boto3

# DynamoDB
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('Users')
table.put_item(Item={'pk': f'USER#{id}', 'sk': 'PROFILE', 'name': name})
response = table.get_item(Key={'pk': f'USER#{id}', 'sk': 'PROFILE'})

# Lambda
lambda_client = boto3.client('lambda')
lambda_client.invoke(FunctionName='processOrder',
    InvocationType='Event', Payload=json.dumps({'orderId': order_id}))

# SNS
sns = boto3.client('sns')
sns.publish(TopicArn=topic_arn, Message=json.dumps(payload))

Credentials (never hardcode)

Anti-Patterns

Production Checklist

• IAM roles with least privilege
• No hardcoded credentials
• SDK retry configuration (maxAttempts)
• CloudWatch alarms on error metrics
• VPC endpoints for S3/DynamoDB (saves NAT costs)
• Resource tagging for cost allocation