TestMu 2025 - The Largest Virtual Software Testing Conference is Back | 50,000+ Attendees

How to use put_record_batch method in localstack

Best Python code snippet using localstack_python

test_firehose_dal.py

...29    firehose_service.update_failed_by_error_code(kinesis_response)30    assert firehose_service.failed_by_error_code == {RANDOM_ERROR_CODE: 1}31def test_put_records_happy_flow():32    firehose = FirehoseDal(stream_name="stream_name", account_id="random")33    assert firehose.put_record_batch(records=[{"id": 1}]) == 134def test_put_records_happy_flow_with_non_default_batch_size():35    firehose = FirehoseDal(stream_name="stream_name", batch_size=1, account_id="random")36    assert firehose.put_record_batch(records=[{"id": 1}]) == 137def test_put_records_with_record_too_big():38    firehose = FirehoseDal(stream_name="stream_name", account_id="random")39    assert firehose.put_record_batch(records=[{"id": ("x" * 2048000)}]) == 040def test_put_records_with_record_invalid():41    firehose = FirehoseDal(stream_name="stream_name", account_id="random")42    assert firehose.put_record_batch(records=[Exception()]) == 0  # noqa43def test_put_records_error_on_first_try_success_in_second_try(monkeypatch):44    monkeypatch.setattr(45        FirehoseDal,46        "get_boto_client",47        lambda x, y: MockFirehoseBotoClientRetryWithError(),48    )49    firehose = FirehoseDal(stream_name="stream_name", account_id="random")50    assert firehose.put_record_batch(records=[{"id": 1}]) == 151def test_put_records_exception_on_first_try_success_in_second_try(monkeypatch):52    client = MockFirehoseBotoClientException()53    monkeypatch.setattr(FirehoseDal, "get_boto_client", lambda x, y: client)54    firehose = FirehoseDal(stream_name="stream_name", account_id="random")55    assert firehose.put_record_batch(records=[{"id": 1}]) == 156def test_put_records_dont_retry_to_many_times(monkeypatch):57    monkeypatch.setattr(58        FirehoseDal,59        "get_boto_client",60        lambda x, y: MockFirehoseBotoClientRetryWithError(3),61    )62    firehose = FirehoseDal(63        stream_name="stream_name", max_retry_count=2, account_id="random"64    )65    assert firehose.put_record_batch(records=[{"id": 1}]) == 066@pytest.mark.skip_firehose_dal_mock67@mock_sts68def test_get_boto_client_in_china_should_connect_to_global(monkeypatch):69    monkeypatch.setenv("LUMIGO_LOGS_EDGE_AWS_ACCESS_KEY_ID", "key1")70    monkeypatch.setenv("LUMIGO_LOGS_EDGE_AWS_SECRET_ACCESS_KEY", "secret1")71    monkeypatch.setenv("AWS_REGION", "cn-northwest-1")72    client = FirehoseDal.get_boto_client("111111111111", "unittests")73    assert client.meta.region_name == "us-west-2"74class MockFirehoseBotoClientRetryWithException:75    retry = 176    def put_record_batch(self, DeliveryStreamName: str, Records: List[dict]):77        if self.retry > 0:78            self.retry -= 179            raise Exception()80        return {81            "FailedPutCount": 0,82            "RequestResponses": [{"RecordId": "1"} for _ in Records],83        }84class MockFirehoseBotoClientRetryWithError:85    def __init__(self, retry=1):86        self.retry = retry87    def put_record_batch(self, DeliveryStreamName: str, Records: List[dict]):88        if self.retry > 0:89            self.retry -= 190            return {91                "FailedPutCount": 0,92                "RequestResponses": [93                    {94                        "ErrorCode": "ServiceUnavailableException",95                        "ErrorMessage": "ServiceUnavailableException",96                    }97                    for _ in Records98                ],99            }100        return {101            "FailedPutCount": 0,102            "RequestResponses": [{"RecordId": "1"} for _ in Records],103        }104class MockFirehoseBotoClientException:105    retry = 1106    records: List = []107    def put_record_batch(self, DeliveryStreamName: str, Records: List[dict]):108        if self.retry > 0:109            self.retry -= 1110            raise Exception()111        self.records.extend(Records)112        return {113            "FailedPutCount": 0,114            "RequestResponses": [{"RecordId": "1"} for _ in Records],115        }116@pytest.fixture117def kinesis_success_item() -> dict:118    return {"RecordId": RANDOM_RECORD_ID}119@pytest.fixture120def kinesis_failed_item() -> dict:121    return {"ErrorCode": RANDOM_ERROR_CODE}

lambdacode.py

Source: lambdacode.py

...44        }45        records.append(logEvent)46        if len(records) > 499:47            print("Records are greater than 499 lets batch and send to firehose %s " % len(records))48            response = firehose.put_record_batch(49                DeliveryStreamName = os.environ['DELIVERY_STREAM_NAME'],50                Records = records51            )52            print("Response from firehose put_record_batch in message process loop is: %s" % response)53            records = []54    if len(records) > 0:55        response = firehose.put_record_batch(56            DeliveryStreamName = os.environ['DELIVERY_STREAM_NAME'],57            Records = records58        )59        print("Response from firehose put_record_batch after message process loop is: %s" % response)60def getEniId(message):61    records = message.split()62    if len(records) > 0:63        return records[2]64    else:65        return None66def getSgIds(netInfObj):67    sgroups = netInfObj.groups68    if len(sgroups) > 0:69        tempsgroupids = []...

data_lake.py

Source: data_lake.py

...42                raise e43            else:44                warn(e)45                return False, str(e)46    def put_record_batch(self, items: list) -> Tuple[bool, str]:47        try:48            response = self.client.put_record_batch(49                DeliveryStreamName=self.DELIVERY_STREAM_NAME,50                Records=[{'Data': self.convert_to_byte(item)} for item in items],51            )52            return True, None53        except Exception as e:54            if settings.DEBUG:55                raise e56            else:57                warn(e)...

Blogs

Check out the latest blogs from LambdaTest on this topic:

How To Use CSS Contain Property To Optimize Browsers

How to increase and maintain team motivation

The best agile teams are built from people who work together as one unit, where each team member has both the technical and the personal skills to allow the team to become self-organized, cross-functional, and self-motivated. These are all big words that I hear in almost every agile project. Still, the criteria to make a fantastic agile team are practically impossible to achieve without one major factor: motivation towards a common goal.

A Complete Guide To CSS Houdini

As a developer, checking the cross browser compatibility of your CSS properties is of utmost importance when building your website. I have often found myself excited to use a CSS feature only to discover that it’s still not supported on all browsers. Even if it is supported, the feature might be experimental and not work consistently across all browsers. Ask any front-end developer about using a CSS feature whose support is still in the experimental phase in most prominent web browsers. ????

Get A Seamless Digital Experience With #LambdaTestYourBusiness????

The holidays are just around the corner, and with Christmas and New Year celebrations coming up, everyone is busy preparing for the festivities! And during this busy time of year, LambdaTest also prepped something special for our beloved developers and testers – #LambdaTestYourBusiness

Digital transformation – part 1 – basics

Digital transformation - part 1 - basics

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.