dynamodb_conditional_write.py

"""
Conditional write for idempotent-ish processing (e.g., SQS worker with retries).

If the item already exists with the same idempotency key, ConditionalCheckFailedException
is raised — treat as success (duplicate delivery).
"""

import boto3
from botocore.exceptions import ClientError

dynamodb = boto3.resource("dynamodb", region_name="us-east-1")


def record_processed_message(table_name: str, message_id: str, payload: dict) -> bool:
    """
    Returns True if this message_id was recorded for the first time.
    Returns False if another worker already recorded it (duplicate).
    """
    table = dynamodb.Table(table_name)
    try:
        table.put_item(
            Item={"pk": f"MSG#{message_id}", "data": payload},
            ConditionExpression="attribute_not_exists(pk)",
        )
        return True
    except ClientError as e:
        if e.response["Error"]["Code"] == "ConditionalCheckFailedException":
            return False
        raise