boto3 DynamoDB idempotency and SQS queue tested offline with moto: working versions (tested 2026-10-15)

boto3==1.43.114
moto==5.2.4
pytest==9.1.1

import json
from decimal import Decimal

import boto3
from botocore.exceptions import ClientError

TABLE = "idempotency"
QUEUE = "orders"


def create_resources(region="us-east-1"):
    ddb = boto3.resource("dynamodb", region_name=region)
    table = ddb.create_table(
        TableName=TABLE,
        KeySchema=[{"AttributeName": "pk", "KeyType": "HASH"}],
        AttributeDefinitions=[{"AttributeName": "pk", "AttributeType": "S"}],
        BillingMode="PAY_PER_REQUEST",
    )
    sqs = boto3.client("sqs", region_name=region)
    url = sqs.create_queue(QueueName=QUEUE)["QueueUrl"]
    return table, sqs, url


class OrderStore:
    def __init__(self, table, sqs, queue_url):
        self.table = table
        self.sqs = sqs
        self.url = queue_url

    def record_once(self, key: str, amount: Decimal) -> bool:
        """Return True on first write, False if the key already exists."""
        try:
            self.table.put_item(
                Item={"pk": key, "amount": amount},
                ConditionExpression="attribute_not_exists(pk)",
            )
            return True
        except ClientError as e:
            if e.response["Error"]["Code"] == "ConditionalCheckFailedException":
                return False
            raise

    def enqueue(self, payload: dict) -> None:
        self.sqs.send_message(QueueUrl=self.url, MessageBody=json.dumps(payload))

    def drain(self) -> list[dict]:
        out = []
        # An empty queue returns no "Messages" key at all, not an empty list.
        resp = self.sqs.receive_message(QueueUrl=self.url, MaxNumberOfMessages=10)
        for m in resp.get("Messages", []):
            out.append(json.loads(m["Body"]))
            self.sqs.delete_message(QueueUrl=self.url, ReceiptHandle=m["ReceiptHandle"])
        return out

from decimal import Decimal

import pytest
from moto import mock_aws

from store import OrderStore, create_resources


@pytest.fixture(autouse=True)
def aws(monkeypatch):
    # Fake credentials so boto3 never reads real ~/.aws config or hits the network.
    monkeypatch.setenv("AWS_ACCESS_KEY_ID", "testing")
    monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "testing")
    monkeypatch.setenv("AWS_DEFAULT_REGION", "us-east-1")
    # Clients must be created while mock_aws is active.
    with mock_aws():
        yield


@pytest.fixture
def store():
    table, sqs, url = create_resources()
    return OrderStore(table, sqs, url)


def test_conditional_put_is_idempotent(store):
    assert store.record_once("order-1", Decimal("9.99")) is True
    assert store.record_once("order-1", Decimal("1.00")) is False
    item = store.table.get_item(Key={"pk": "order-1"})["Item"]
    assert item["amount"] == Decimal("9.99")


def test_floats_are_rejected_use_decimal(store):
    # boto3's serializer refuses float; amounts must be Decimal.
    with pytest.raises(TypeError):
        store.record_once("order-2", 9.99)


def test_sqs_roundtrip_and_delete(store):
    store.enqueue({"id": 1})
    store.enqueue({"id": 2})
    assert sorted(m["id"] for m in store.drain()) == [1, 2]
    # Messages were deleted, so a second drain sees nothing.
    assert store.drain() == []


def test_empty_queue_returns_empty_list(store):
    assert store.drain() == []

[pytest]
pythonpath = .