Function writer_thread

samples/stream_consumer.py:47–91

()

        time.sleep(0.5)
    raise TimeoutError(f"Table {table_name} did not become ACTIVE within {timeout}s")
def writer_thread() -> None:
    client = make_client()

    # Phase 1: Insert items
    for i in range(1, ITEM_COUNT + 1):
        client.put_item(
            TableName=TABLE_NAME,
            Item={
                "pk": {"S": f"item-{i}"},
                "version": {"N": "1"},
                "data": {"S": f"initial-value-{i}"},
            },
        )
        log("WRITER", f"INSERT  pk=item-{i}  version=1  data=initial-value-{i}")
        time.sleep(0.3)

    time.sleep(1)

    # Phase 2: Update items
    for i in range(1, ITEM_COUNT + 1):
        client.update_item(
            TableName=TABLE_NAME,
            Key={"pk": {"S": f"item-{i}"}},
            UpdateExpression="SET version = :v, #d = :d",
            ExpressionAttributeNames={"#d": "data"},
            ExpressionAttributeValues={
                ":v": {"N": "2"},
                ":d": {"S": f"updated-value-{i}"},
            },
        )
        log("WRITER", f"UPDATE  pk=item-{i}  version=2  data=updated-value-{i}")
        time.sleep(0.3)

    time.sleep(1)

    # Phase 3: Delete items
    for i in range(1, ITEM_COUNT + 1):
        client.delete_item(
            TableName=TABLE_NAME,
            Key={"pk": {"S": f"item-{i}"}},
        )
        log("WRITER", f"DELETE  pk=item-{i}")
        time.sleep(0.3)

    log("WRITER", "All writes complete. Waiting for poller to catch up...")
def format_image(image: dict | None) -> str:
    if not image:
        return "{}"