How to use publish_batch method in localstack

Best Python code snippet using localstack_python

test_consumer.py

Source:test_consumer.py

...26async def test_consume(stream: str, consumer: Consumer, producer: Producer) -> None:27    captured: list[bytes] = []28    await consumer.subscribe(stream, callback=captured.append)29    assert await producer.publish(stream, b"one") == 130    assert await producer.publish_batch(stream, [b"two", b"three"]) == [2, 3]31    await wait_for(lambda: len(captured) >= 3)32    assert captured == [b"one", b"two", b"three"]33async def test_offset_type_first(stream: str, consumer: Consumer, producer: Producer) -> None:34    captured: list[bytes] = []35    await consumer.subscribe(36        stream,37        callback=captured.append,38        offset_type=OffsetType.FIRST,39    )40    messages = [str(i).encode() for i in range(1, 11)]41    await producer.publish_batch(stream, messages)42    await wait_for(lambda: len(captured) >= 10)43    assert captured == messages44async def test_offset_type_offset(stream: str, consumer: Consumer, producer: Producer) -> None:45    captured: list[bytes] = []46    await consumer.subscribe(47        stream,48        callback=captured.append,49        offset_type=OffsetType.OFFSET,50        offset=7,51    )52    messages = [str(i).encode() for i in range(1, 11)]53    await producer.publish_batch(stream, messages)54    await wait_for(lambda: len(captured) >= 3)55    assert captured == messages[7:]56async def test_offset_type_last(stream: str, consumer: Consumer, producer: Producer) -> None:57    messages = [str(i).encode() for i in range(1, 5_000)]58    await producer.publish_batch(stream, messages)59    captured: list[bytes] = []60    await consumer.subscribe(61        stream,62        callback=captured.append,63        offset_type=OffsetType.LAST,64        subscriber_name="test-subscriber",65    )66    await wait_for(lambda: len(captured) > 0 and captured[-1] == b"4999")67    assert len(captured) < len(messages)68async def test_offset_type_timestamp(stream: str, consumer: Consumer, producer: Producer) -> None:69    messages = [str(i).encode() for i in range(1, 5_000)]70    await producer.publish_batch(stream, messages)71    # mark time in between message batches72    now = int(time.time() * 1000)73    messages = [str(i).encode() for i in range(5_000, 5_100)]74    await producer.publish_batch(stream, messages)75    captured: list[bytes] = []76    await consumer.subscribe(stream, callback=captured.append, offset_type=OffsetType.TIMESTAMP, offset=now)77    await wait_for(lambda: len(captured) > 0 and captured[0] >= b"5000")78async def test_offset_type_next(stream: str, consumer: Consumer, producer: Producer) -> None:79    messages = [str(i).encode() for i in range(1, 11)]80    await producer.publish_batch(stream, messages)81    captured: list[bytes] = []82    await consumer.subscribe(83        stream,84        callback=captured.append,85        offset_type=OffsetType.NEXT,86        subscriber_name="test-subscriber",87    )88    await producer.publish(stream, b"11")89    await wait_for(lambda: len(captured) > 0)90    assert captured == [b"11"]91async def test_consume_with_resubscribe(stream: str, consumer: Consumer, producer: Producer) -> None:92    captured: list[bytes] = []93    subscriber_name = await consumer.subscribe(stream, callback=captured.append)94    await producer.publish(stream, b"one")...

batch_test.py

Source:batch_test.py

...18        now_time = str(datetime.now()).replace(" ", "_")19        project_batch = create_batch(client, project.project, f"Batch-{now_time}")20        assert isinstance(project_batch, ProjectBatch)21        assert project_batch.status == "open"22    def test_publish_batch(self, client: IAC.InputApiClient):23        projects = get_projects(client=client)24        project = self.filter_batch_project(projects)25        now_time = str(datetime.now()).replace(" ", "_")26        created_batch = create_batch(client, project.project, f"Batch-{now_time}")27        published_batch = publish_batch(client, project.project, created_batch.batch)...

publisher.py

Source:publisher.py

2import time3from os.path import basename4import glob5import boto36def publish_batch(client, streamname, batch):7    response = client.put_record_batch(DeliveryStreamName=streamname, Records=batch)8    if response["FailedPutCount"] > 0:9        print(f"{response['FailedPutCount']} failed, retrying")10        for i, obj in enumerate(response["RequestResponses"]):11            if "RecordId" in obj:12                del batch[i]13        time.sleep(1)14        publish_batch(client, streamname, batch)15    else:16        batch.clear()17if __name__ == "__main__":18    sess = boto3.Session(region_name="us-east-1",)19    client = sess.client("firehose")20    batchsize = 50021    res = client.list_delivery_streams()22    streamname, *b = [23        i for i in res["DeliveryStreamNames"] if i.startswith("streaming_submissions-")24    ]25    batch = []26    for filename in glob.glob(sys.argv[1]):27        with open(filename) as csvf:28            for i, row in enumerate(csvf.readlines()):29                if len(batch) >= batchsize:30                    publish_batch(client, streamname, batch)31                batch.append({"Data": basename(filename) + " " + row})32            else:33                publish_batch(client, streamname, batch)34                print(35                    f"Published {i + 1} records to {streamname} from {filename} in batches of {batchsize}"...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.