Function main

samples/stream_consumer.py:182–221

()

    log("POLLER", f"Done. Total records seen: {records_seen}")
def main() -> None:
    client = make_client()

    log("MAIN  ", f"Creating table {TABLE_NAME} with streams (NEW_AND_OLD_IMAGES)...")
    client.create_table(
        TableName=TABLE_NAME,
        AttributeDefinitions=[{"AttributeName": "pk", "AttributeType": "S"}],
        KeySchema=[{"AttributeName": "pk", "KeyType": "HASH"}],
        BillingMode="PAY_PER_REQUEST",
        StreamSpecification={
            "StreamEnabled": True,
            "StreamViewType": "NEW_AND_OLD_IMAGES",
        },
    )
    wait_for_active(client, TABLE_NAME)
    log("MAIN  ", "Table ACTIVE.")

    try:
        poller = threading.Thread(target=poller_thread, name="poller", daemon=True)
        writer = threading.Thread(target=writer_thread, name="writer")

        poller.start()
        time.sleep(1)  # Let poller discover shards before writes begin
        writer.start()

        writer.join()

        # Give poller time to drain remaining records
        log("MAIN  ", "Writer done. Giving poller 5s to drain...")
        time.sleep(5)
        stop_poller.set()
        poller.join(timeout=3)

        log("MAIN  ", "Demo complete.")
    finally:
        log("MAIN  ", f"Deleting table {TABLE_NAME}...")
        try:
            client.delete_table(TableName=TABLE_NAME)
        except Exception:
            pass
if __name__ == "__main__":
    main()