Streaming records with and without bloop

0_by_hand.py

import boto3

dynamodb = boto3.client("dynamodb")
streams = boto3.client("dynamodbstreams")

# The rest of this is a lovecraftian horror I don't want to write right now.

# DynamoDB Stream's documentation only manages a working example without KCL
# by making half a dozen assumptions and forgoing all error handling.

1_with_bloop.py

# Records have real User objects, not 4-deep nested dicts.
# Handles shard rollover, expiration, iterator refresh, multiple shards...
import arrow


def process(record):
    print("user's email updated to {} {} ".format(
            record["new"].email, record["meta"]["created_at"].humanize()))


stream = engine.stream(User, "trim_horizon")


next_heartbeat = arrow.now()
while True:
    record = next(stream)
    if record: process(record)
    if arrow.now() > next_heartbeat:
        next_heartbeat = arrow.now().replace(minutes=12)
        stream.heartbeat()