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"]])

-- One-time setup: pgcrypto provides digest() for SHA-256 hashing
CREATE EXTENSION IF NOT EXISTS pgcrypto;

-- Business table (mutable current state)
CREATE TABLE accounts (
    account_id TEXT PRIMARY KEY,
    balance    NUMERIC(15,2) NOT NULL DEFAULT 0
);

-- Append-only, hash-chained journal: every change becomes a new row.
-- Each row's hash covers the previous row's hash, so altering any
-- historical row breaks the chain from that point onward.
CREATE TABLE audit_log (
    seq        BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    changed_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    table_name TEXT        NOT NULL,
    operation  TEXT        NOT NULL,   -- INSERT / UPDATE / DELETE
    row_data   JSONB       NOT NULL,   -- full image of the changed row
    prev_hash  BYTEA,                  -- row_hash of the previous journal entry
    row_hash   BYTEA       NOT NULL    -- sha256(prev_hash || row_data)
);

-- Trigger that appends a hash-chained entry for every write to accounts
CREATE OR REPLACE FUNCTION append_audit() RETURNS TRIGGER AS $$
DECLARE
    last_hash BYTEA;
    payload   JSONB := to_jsonb(COALESCE(NEW, OLD));  -- OLD on DELETE
BEGIN
    SELECT row_hash INTO last_hash
      FROM audit_log ORDER BY seq DESC LIMIT 1;       -- NULL for the first row

    INSERT INTO audit_log (table_name, operation, row_data, prev_hash, row_hash)
    VALUES (
        TG_TABLE_NAME, TG_OP, payload, last_hash,
        digest(COALESCE(last_hash, ''::bytea) || convert_to(payload::text, 'UTF8'), 'sha256')
    );
    RETURN NULL;  -- AFTER trigger: return value is ignored
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER accounts_audit
    AFTER INSERT OR UPDATE OR DELETE ON accounts
    FOR EACH ROW EXECUTE FUNCTION append_audit();

-- Enforce append-only: the application role may read and append, never mutate history
GRANT SELECT, INSERT ON audit_log TO app_role;
REVOKE UPDATE, DELETE, TRUNCATE ON audit_log FROM app_role;

import psycopg  # psycopg 3

conn = psycopg.connect(
    host="my-aurora-cluster.cluster-xyz.us-east-1.rds.amazonaws.com",
    dbname="appdb", user="app_role", password="S3cr3tP@ss!",
)

def transfer(conn, from_id, to_id, amount):
    # A single ACID transaction. The AFTER trigger appends a hash-chained
    # journal row for each UPDATE automatically - no application code needed.
    with conn.transaction():
        with conn.cursor() as cur:
            cur.execute(
                "UPDATE accounts SET balance = balance - %s WHERE account_id = %s",
                (amount, from_id),
            )
            cur.execute(
                "UPDATE accounts SET balance = balance + %s WHERE account_id = %s",
                (amount, to_id),
            )

transfer(conn, "acct-001", "acct-002", 500)

# Full immutable history of an account: every revision, not just current state
with conn.cursor() as cur:
    cur.execute(
        """
        SELECT seq, changed_at, operation, row_data
        FROM audit_log
        WHERE row_data->>'account_id' = %s
        ORDER BY seq
        """,
        ("acct-001",),
    )
    for row in cur.fetchall():
        print(row)

# Tamper-evidence check: recompute every hash and confirm the chain is intact
with conn.cursor() as cur:
    cur.execute(
        """
        SELECT bool_and(
            row_hash = digest(
                COALESCE(prev_hash, ''::bytea) || convert_to(row_data::text, 'UTF8'),
                'sha256'
            )
        ) AS chain_intact
        FROM audit_log
        """
    )
    print("chain intact:", cur.fetchone()[0])