Managed Databases

# Create a PostgreSQL RDS instance (single-AZ for dev)
aws rds create-db-instance \
  --db-instance-identifier my-postgres \
  --db-instance-class db.t3.micro \
  --engine postgres \
  --engine-version 16.2 \
  --master-username admin \
  --master-user-password "S3cr3tP@ss!" \
  --allocated-storage 20 \
  --db-subnet-group-name my-db-subnet-group \
  --vpc-security-group-ids sg-0123456789abcdef0 \
  --no-publicly-accessible

# Wait until it's available (takes ~5 minutes)
aws rds wait db-instance-available \
  --db-instance-identifier my-postgres

# Get the endpoint
aws rds describe-db-instances \
  --db-instance-identifier my-postgres \
  --query 'DBInstances[0].Endpoint.Address' --output text

# Create an Aurora PostgreSQL Serverless v2 cluster
aws rds create-db-cluster \
  --db-cluster-identifier my-aurora-cluster \
  --engine aurora-postgresql \
  --engine-version 15.4 \
  --serverless-v2-scaling-configuration MinCapacity=0.5,MaxCapacity=8 \
  --database-name appdb \
  --master-username admin \
  --master-user-password "S3cr3tP@ss!" \
  --db-subnet-group-name my-db-subnet-group \
  --vpc-security-group-ids sg-0123456789abcdef0

# Add a Serverless v2 writer instance to the cluster
aws rds create-db-instance \
  --db-instance-identifier my-aurora-writer \
  --db-instance-class db.serverless \
  --engine aurora-postgresql \
  --db-cluster-identifier my-aurora-cluster

# Get the cluster endpoint (writer)
aws rds describe-db-clusters \
  --db-cluster-identifier my-aurora-cluster \
  --query 'DBClusters[0].Endpoint' --output text

# Create a DynamoDB table (orders with composite key)
aws dynamodb create-table \
  --table-name Orders \
  --attribute-definitions \
    AttributeName=userId,AttributeType=S \
    AttributeName=orderId,AttributeType=S \
  --key-schema \
    AttributeName=userId,KeyType=HASH \
    AttributeName=orderId,KeyType=RANGE \
  --billing-mode PAY_PER_REQUEST

# Put an item
aws dynamodb put-item \
  --table-name Orders \
  --item '{
    "userId":  {"S": "user-123"},
    "orderId": {"S": "ord-456"},
    "status":  {"S": "pending"},
    "total":   {"N": "49.99"}
  }'

# Query all orders for a user (efficient - uses the partition key)
aws dynamodb query \
  --table-name Orders \
  --key-condition-expression "userId = :uid" \
  --expression-attribute-values '{":uid": {"S": "user-123"}}'

import boto3
from boto3.dynamodb.conditions import Key

dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table("Orders")

# Query orders for a user
response = table.query(
    KeyConditionExpression=Key("userId").eq("user-123")
)
orders = response["Items"]

# Put a new order
table.put_item(Item={
    "userId":  "user-123",
    "orderId": "ord-789",
    "status":  "pending",
    "total":   49.99,
})

# Update status
table.update_item(
    Key={"userId": "user-123", "orderId": "ord-789"},
    UpdateExpression="SET #s = :status",
    ExpressionAttributeNames={"#s": "status"},
    ExpressionAttributeValues={":status": "shipped"},
)

# Create a DocumentDB cluster (MongoDB-compatible)
aws docdb create-db-cluster \
  --db-cluster-identifier my-docdb-cluster \
  --engine docdb \
  --engine-version 5.0.0 \
  --master-username admin \
  --master-user-password "S3cr3tP@ss!" \
  --db-subnet-group-name my-db-subnet-group \
  --vpc-security-group-ids sg-0123456789abcdef0

# Add a primary instance to the cluster
aws docdb create-db-instance \
  --db-instance-identifier my-docdb-instance \
  --db-instance-class db.r6g.large \
  --engine docdb \
  --db-cluster-identifier my-docdb-cluster

# Connect and query with pymongo (DocumentDB uses TLS by default)
# Download the CA bundle first:
# wget https://truststore.pki.rds.amazonaws.com/global/global-bundle.pem

import pymongo

# Connect to DocumentDB (TLS required)
# Use host/username/password kwargs - avoid embedding special chars in the URI
client = pymongo.MongoClient(
    host="my-docdb-cluster.cluster-xyz.us-east-1.docdb.amazonaws.com",
    port=27017,
    username="admin",
    password="S3cr3tP@ss!",
    tls=True,
    tlsCAFile="global-bundle.pem",
    retryWrites=False,  # DocumentDB does not support retryWrites
)

db = client["catalog"]
products = db["products"]

# Insert a document (flexible schema - no migration needed)
products.insert_one({
    "sku": "WIDGET-42",
    "name": "Blue Widget",
    "tags": ["sale", "featured"],
    "specs": {"weight_kg": 0.5, "color": "blue"},
    "price": 19.99,
})

# Rich query: find all featured products under $25 sorted by price
results = products.find(
    {"tags": "featured", "price": {"$lt": 25}},
    sort=[("price", pymongo.ASCENDING)],
)

# Aggregation pipeline: average price by tag
pipeline = [
    {"$unwind": "$tags"},
    {"$group": {"_id": "$tags", "avg_price": {"$avg": "$price"}}},
    {"$sort": {"avg_price": -1}},
]
for bucket in products.aggregate(pipeline):
    print(bucket)

# Create a Neptune cluster (graph database)
aws neptune create-db-cluster \
  --db-cluster-identifier my-neptune-cluster \
  --engine neptune \
  --engine-version 1.3.1.0 \
  --db-subnet-group-name my-db-subnet-group \
  --vpc-security-group-ids sg-0123456789abcdef0 \
  --storage-encrypted

# Add a primary instance to the cluster
aws neptune create-db-instance \
  --db-instance-identifier my-neptune-instance \
  --db-instance-class db.r6g.large \
  --engine neptune \
  --db-cluster-identifier my-neptune-cluster

# Get the cluster endpoint
aws neptune describe-db-clusters \
  --db-cluster-identifier my-neptune-cluster \
  --query 'DBClusters[0].Endpoint' --output text

# Neptune supports openCypher (graph query language)
# Use the neptune-graph SDK or boto3 neptune-graph client

import boto3

# For Neptune Analytics (in-memory graph - fast traversals)
client = boto3.client("neptune-graph", region_name="us-east-1")

# openCypher query: find all fraud rings (accounts connected through shared devices)
query = """
MATCH (a1:Account)-[:USED]->(d:Device)<-[:USED]-(a2:Account)
WHERE a1 <> a2
  AND a1.flagged = true
RETURN a2.accountId AS suspicious, d.deviceId AS sharedDevice
LIMIT 100
"""

response = client.execute_query(
    graphIdentifier="my-neptune-graph-id",
    queryString=query,
    language="OPEN_CYPHER",
)

# openCypher query: shortest path between two users (social graph)
path_query = """
MATCH p = shortestPath(
  (u1:User {id: 'user-001'})-[*]-(u2:User {id: 'user-999'})
)
RETURN length(p) AS hops, [n IN nodes(p) | n.id] AS path
"""

# Create a Timestream database and table
aws timestream-write create-database \
  --database-name IoTMetrics

aws timestream-write create-table \
  --database-name IoTMetrics \
  --table-name SensorReadings \
  --retention-properties '{
    "MemoryStoreRetentionPeriodInHours": 24,
    "MagneticStoreRetentionPeriodInDays": 365
  }'

import boto3
from datetime import datetime, timezone

write_client = boto3.client("timestream-write", region_name="us-east-1")
query_client = boto3.client("timestream-query", region_name="us-east-1")

# Write a batch of sensor readings
now_ms = str(int(datetime.now(timezone.utc).timestamp() * 1000))

write_client.write_records(
    DatabaseName="IoTMetrics",
    TableName="SensorReadings",
    Records=[
        {
            "Dimensions": [
                {"Name": "device_id",  "Value": "sensor-42"},
                {"Name": "location",   "Value": "warehouse-A"},
            ],
            "MeasureName":  "temperature",
            "MeasureValue": "23.7",
            "MeasureValueType": "DOUBLE",
            "Time": now_ms,
        },
    ],
)

# Query: average temperature per device over the last hour
# Uses Timestream SQL with built-in time functions
sql = """
SELECT device_id,
       bin(time, 5m)                AS five_min_window,
       avg(measure_value::double)   AS avg_temp,
       max(measure_value::double)   AS peak_temp
FROM "IoTMetrics"."SensorReadings"
WHERE measure_name = 'temperature'
  AND time >= ago(1h)
GROUP BY device_id, bin(time, 5m)
ORDER BY five_min_window DESC
"""

response = query_client.query(QueryString=sql)
for row in response["Rows"]:
    print([col.get("ScalarValue") for col in row["Data"]])

# Create a QLDB ledger
aws qldb create-ledger \
  --name my-finance-ledger \
  --permissions-mode STANDARD \
  --deletion-protection  # prevents accidental deletion

# Describe the ledger (wait until ACTIVE)
aws qldb describe-ledger --name my-finance-ledger

from datetime import datetime, timezone
from pyqldb.driver.qldb_driver import QldbDriver

driver = QldbDriver(ledger_name="my-finance-ledger")

def create_tables(transaction_executor):
    transaction_executor.execute_statement(
        "CREATE TABLE Accounts"
    )
    transaction_executor.execute_statement(
        "CREATE INDEX ON Accounts (accountId)"
    )

def record_transfer(transaction_executor, from_id, to_id, amount):
    # QLDB uses PartiQL (SQL-compatible)
    # Every INSERT/UPDATE is appended to the immutable journal
    now_iso = datetime.now(timezone.utc).isoformat()
    transaction_executor.execute_statement(
        "UPDATE Accounts SET balance = balance - ? WHERE accountId = ?",
        amount, from_id,
    )
    transaction_executor.execute_statement(
        "UPDATE Accounts SET balance = balance + ? WHERE accountId = ?",
        amount, to_id,
    )
    transaction_executor.execute_statement(
        "INSERT INTO Transfers VALUE {'from': ?, 'to': ?, 'amount': ?, 'ts': ?}",
        from_id, to_id, amount, now_iso,
    )

# Execute as a single ACID transaction
driver.execute_lambda(lambda txn: record_transfer(txn, "acct-001", "acct-002", 500))

# Query the full history of an account (immutable audit trail)
def get_history(transaction_executor, account_id):
    return transaction_executor.execute_statement(
        "SELECT * FROM history(Accounts) WHERE data.accountId = ?",
        account_id,
    )

history = driver.execute_lambda(lambda txn: get_history(txn, "acct-001"))