10.3SQS Polling Helpers

import time
import boto3

def poll_sqs(queue_url: str, timeout: int = 10) -> dict | None:
    """Poll an SQS queue for a single message with timeout."""
    sqs = boto3.client('sqs', region_name='us-east-1')
    deadline = time.time() + timeout

    while time.time() < deadline:
        response = sqs.receive_message(
            QueueUrl=queue_url,
            MaxNumberOfMessages=1,
            WaitTimeSeconds=2,
        )
        messages = response.get("Messages", [])
        if messages:
            # Delete the message so it does not interfere with other tests
            sqs.delete_message(
                QueueUrl=queue_url,
                ReceiptHandle=messages[0]["ReceiptHandle"]
            )
            return messages[0]

    return None

def poll_sqs_all(queue_url: str, timeout: int = 10) -> list[dict]:
    """Poll all available messages from an SQS queue."""
    sqs = boto3.client('sqs', region_name='us-east-1')
    all_messages = []
    deadline = time.time() + timeout

    while time.time() < deadline:
        response = sqs.receive_message(
            QueueUrl=queue_url,
            MaxNumberOfMessages=10,
            WaitTimeSeconds=1,
        )
        messages = response.get("Messages", [])
        if not messages:
            break
        all_messages.extend(messages)
        for msg in messages:
            sqs.delete_message(
                QueueUrl=queue_url,
                ReceiptHandle=msg["ReceiptHandle"]
            )

    return all_messages