Example of a possible Bloop extension to do cross-region replication with AWS Lambda in 30 lines

bloop_lambda_replication.py

from bloop import Engine
from bloop.ext._lambda import unpack
from boto3 import Session

from my_models_script import MyModel

primary = Engine()
session = Session(region_name="us-east-1")
backup = Engine(
    dynamodb=session.client("dynamodb"),
    dynamodbstreams=session.client("dynamodbstreams"))
primary.bind(MyModel)
backup.bind(MyModel)


def replicate(record):
    unpack(engine, record)
    change_type = record["meta"]["event"]["type"]
    obj = record["key"] or record["new"] or record["old"]

    if change_type in {"insert", "modify"}:
        backup.save(obj)
    elif change_type == "remove":
        backup.delete(obj)
    else:
        raise ValueError(f"Unknown change type '{change_type}'")


def handler(event, context):
    for record in event.get("Records", []):
        replicate(record)